Merge remote-tracking branch 'origin/bugfix/fix_nullable_enum_avro_se…

…r' into bugfix/fix_nullable_enum_avro_ser

.github/workflows/backend_tests.yml

@@ -28,7 +28,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 

.github/workflows/branch-deploy.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
       - name: Build

.github/workflows/build-public-image.yml

@@ -27,7 +27,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
       - name: Build

.github/workflows/codeql-analysis.yml

@@ -45,7 +45,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 

.github/workflows/cve_checks.yml

@@ -20,7 +20,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 

.github/workflows/e2e-run.yml

@@ -30,7 +30,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v3
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 
@@ -72,7 +72,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v3
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 

.github/workflows/frontend_tests.yml

@@ -23,7 +23,7 @@ jobs:
 
       - uses: pnpm/[email protected]
         with:
-          version: 9.11.0
+          version: 9.15.0
 
       - name: Install node
         uses: actions/[email protected]

.github/workflows/main.yml

@@ -28,7 +28,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 

.github/workflows/release-serde-api.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: "17"
+          java-version: "21"
           distribution: "zulu"
           cache: "maven"
 

.github/workflows/release.yml

@@ -28,7 +28,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
 

.github/workflows/separate_env_public_create.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v4
         with:
-          java-version: '17'
+          java-version: '21'
           distribution: 'zulu'
           cache: 'maven'
       - name: Build

README.md

@@ -15,7 +15,7 @@ Versatile, fast and lightweight web UI for managing Apache Kafka® clusters.
 
 <p align="center">
     <a href="https://ui.docs.kafbat.io/">Documentation</a> • 
-    <a href="https://ui.docs.kafbat.io/configuration/quick-start">Quick Start</a> • 
+    <a href="https://ui.docs.kafbat.io/quick-start/demo-run">Quick Start</a> • 
     <a href="https://discord.gg/4DWzD7pGE5">Community</a>
     <br/>
     <a href="https://aws.amazon.com/marketplace/pp/{replaceMe}">AWS Marketplace</a>  •

api/Dockerfile

@@ -1,7 +1,7 @@
-# The tag is ignored when a sha is included but the reason to add it are:  
+# The tag is ignored when a sha is included but the reason to add it are:
 # 1. Self Documentation: It is difficult to find out what the expected tag is given a sha alone
 # 2. Helps dependabot during discovery of upgrades
-FROM azul/zulu-openjdk-alpine:17-jre-headless-latest@sha256:af4df00adaec356d092651af50d9e80fd179f96722d267e79acb564aede10fda
+FROM azul/zulu-openjdk-alpine:21.0.5-jre-headless@sha256:842b23baf96980437281b7419af970238b4c1c3109e50beac5299cdc278291d7
 
 RUN apk add --no-cache \
     # snappy codec

api/pom.xml

@@ -97,7 +97,7 @@
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-identity</artifactId>
-            <version>1.13.3</version>
+            <version>1.14.2</version>
             <exclusions>
                 <exclusion>
                     <groupId>io.netty</groupId>

api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java

@@ -1,5 +1,6 @@
 package io.kafbat.ui.client;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.connect.ApiClient;
 import io.kafbat.ui.connect.api.KafkaConnectClientApi;
@@ -14,9 +15,11 @@
 import io.kafbat.ui.exception.KafkaConnectConflictReponseException;
 import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.util.WebClientConfigurator;
+import jakarta.validation.constraints.NotNull;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import javax.annotation.Nullable;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.ResponseEntity;
@@ -58,10 +61,24 @@ private static <T> Flux<T> withRetryOnConflict(Flux<T> publisher) {
 
   private static <T> Mono<T> withBadRequestErrorHandling(Mono<T> publisher) {
     return publisher
-        .onErrorResume(WebClientResponseException.BadRequest.class, e ->
-            Mono.error(new ValidationException("Invalid configuration")))
-        .onErrorResume(WebClientResponseException.InternalServerError.class, e ->
-            Mono.error(new ValidationException("Invalid configuration")));
+        .onErrorResume(WebClientResponseException.BadRequest.class,
+            RetryingKafkaConnectClient::parseConnectErrorMessage)
+        .onErrorResume(WebClientResponseException.InternalServerError.class,
+            RetryingKafkaConnectClient::parseConnectErrorMessage);
+  }
+
+  // Adapted from https://github.com/apache/kafka/blob/a0a501952b6d61f6f273bdb8f842346b51e9dfce/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ErrorMessage.java
+  // Adding the connect runtime dependency for this single class seems excessive
+  private record ErrorMessage(@NotNull @JsonProperty("message") String message) {
+  }
+
+  private static <T> @NotNull Mono<T> parseConnectErrorMessage(WebClientResponseException parseException) {
+    final var errorMessage = parseException.getResponseBodyAs(ErrorMessage.class);
+    return Mono.error(new ValidationException(
+        Objects.requireNonNull(errorMessage,
+                // see https://github.com/apache/kafka/blob/a0a501952b6d61f6f273bdb8f842346b51e9dfce/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/ConnectExceptionMapper.java
+                "This should not happen according to the ConnectExceptionMapper")
+            .message()));
   }
 
   @Override
@@ -176,7 +193,7 @@ public Mono<ResponseEntity<Map<String, ConnectorTopics>>> getConnectorTopicsWith
   }
 
   @Override
-  public Flux<String> getConnectors(String search) throws WebClientResponseException {
+  public Mono<List<String>> getConnectors(String search) throws WebClientResponseException {
     return withRetryOnConflict(super.getConnectors(search));
   }
 

api/src/main/java/io/kafbat/ui/emitter/MessageFilters.java

@@ -52,7 +52,23 @@ public static Predicate<TopicMessageDTO> noop() {
 
   public static Predicate<TopicMessageDTO> containsStringFilter(String string) {
     return msg -> StringUtils.contains(msg.getKey(), string)
-        || StringUtils.contains(msg.getContent(), string);
+        || StringUtils.contains(msg.getContent(), string) || headersContains(msg, string);
+  }
+
+  private static boolean headersContains(TopicMessageDTO msg, String searchString) {
+    final var headers = msg.getHeaders();
+
+    if (headers == null) {
+      return false;
+    }
+
+    for (final var entry : headers.entrySet()) {
+      if (StringUtils.contains(entry.getKey(), searchString) || StringUtils.contains(entry.getValue(), searchString)) {
+        return true;
+      }
+    }
+
+    return false;
   }
 
   public static Predicate<TopicMessageDTO> celScriptFilter(String script) {

api/src/main/java/io/kafbat/ui/model/rbac/Role.java

@@ -1,5 +1,6 @@
 package io.kafbat.ui.model.rbac;
 
+import com.google.common.base.Preconditions;
 import java.util.List;
 import lombok.Data;
 
@@ -12,6 +13,7 @@ public class Role {
   List<Permission> permissions;
 
   public void validate() {
+    Preconditions.checkArgument(!clusters.isEmpty(), "Role clusters cannot be empty");
     permissions.forEach(Permission::transform);
     permissions.forEach(Permission::validate);
   }

api/src/main/java/io/kafbat/ui/serdes/builtin/ProtobufFileSerde.java

@@ -15,6 +15,7 @@
 import com.google.protobuf.StructProto;
 import com.google.protobuf.TimestampProto;
 import com.google.protobuf.TypeProto;
+import com.google.protobuf.TypeRegistry;
 import com.google.protobuf.WrappersProto;
 import com.google.protobuf.util.JsonFormat;
 import com.google.type.ColorProto;
@@ -147,12 +148,18 @@ public boolean canSerialize(String topic, Serde.Target type) {
   @Override
   public Serde.Serializer serializer(String topic, Serde.Target type) {
     var descriptor = descriptorFor(topic, type).orElseThrow();
+    TypeRegistry typeRegistry = TypeRegistry.newBuilder()
+        .add(descriptorPaths.keySet())
+        .build();
+
     return new Serde.Serializer() {
       @SneakyThrows
       @Override
       public byte[] serialize(String input) {
         DynamicMessage.Builder builder = DynamicMessage.newBuilder(descriptor);
-        JsonFormat.parser().merge(input, builder);
+        JsonFormat.parser()
+            .usingTypeRegistry(typeRegistry)
+            .merge(input, builder);
         return builder.build().toByteArray();
       }
     };

api/src/main/java/io/kafbat/ui/service/ConsumerGroupService.java

@@ -149,6 +149,8 @@ private Mono<List<ConsumerGroupDescription>> loadSortedDescriptions(ReactiveAdmi
               case EMPTY -> 3;
               case DEAD -> 4;
               case UNKNOWN -> 5;
+              case ASSIGNING -> 6;
+              case RECONCILING -> 7;
             };
         var comparator = Comparator.comparingInt(statesPriorities);
         yield loadDescriptionsByListings(ac, groups, comparator, pageNum, perPage, sortOrderDto);

api/src/main/java/io/kafbat/ui/service/KafkaConnectService.java

@@ -1,6 +1,5 @@
 package io.kafbat.ui.service;
 
-import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.kafbat.ui.connect.api.KafkaConnectClientApi;
 import io.kafbat.ui.connect.model.ConnectorStatus;
@@ -31,7 +30,6 @@
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
-import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
@@ -106,10 +104,7 @@ public Mono<ConnectorTopics> getConnectorTopics(KafkaCluster cluster, String con
 
   public Flux<String> getConnectorNames(KafkaCluster cluster, String connectName) {
     return api(cluster, connectName)
-        .flux(client -> client.getConnectors(null))
-        // for some reason `getConnectors` method returns the response as a single string
-        .collectList().map(e -> e.get(0))
-        .map(this::parseConnectorsNamesStringToList)
+        .mono(client -> client.getConnectors(null))
         .flatMapMany(Flux::fromIterable);
   }
 
@@ -118,12 +113,6 @@ public Flux<String> getConnectorNamesWithErrorsSuppress(KafkaCluster cluster, St
     return getConnectorNames(cluster, connectName).onErrorComplete();
   }
 
-  @SneakyThrows
-  private List<String> parseConnectorsNamesStringToList(String json) {
-    return objectMapper.readValue(json, new TypeReference<>() {
-    });
-  }
-
   public Mono<ConnectorDTO> createConnector(KafkaCluster cluster, String connectName,
                                             Mono<NewConnectorDTO> connector) {
     return api(cluster, connectName)
@@ -218,22 +207,13 @@ public Mono<Void> deleteConnector(
   public Mono<Void> updateConnectorState(KafkaCluster cluster, String connectName,
                                          String connectorName, ConnectorActionDTO action) {
     return api(cluster, connectName)
-        .mono(client -> {
-          switch (action) {
-            case RESTART:
-              return client.restartConnector(connectorName, false, false);
-            case RESTART_ALL_TASKS:
-              return restartTasks(cluster, connectName, connectorName, task -> true);
-            case RESTART_FAILED_TASKS:
-              return restartTasks(cluster, connectName, connectorName,
-                  t -> t.getStatus().getState() == ConnectorTaskStatusDTO.FAILED);
-            case PAUSE:
-              return client.pauseConnector(connectorName);
-            case RESUME:
-              return client.resumeConnector(connectorName);
-            default:
-              throw new IllegalStateException("Unexpected value: " + action);
-          }
+        .mono(client -> switch (action) {
+          case RESTART -> client.restartConnector(connectorName, false, false);
+          case RESTART_ALL_TASKS -> restartTasks(cluster, connectName, connectorName, task -> true);
+          case RESTART_FAILED_TASKS -> restartTasks(cluster, connectName, connectorName,
+              t -> t.getStatus().getState() == ConnectorTaskStatusDTO.FAILED);
+          case PAUSE -> client.pauseConnector(connectorName);
+          case RESUME -> client.resumeConnector(connectorName);
         });
   }
 

api/src/main/java/io/kafbat/ui/service/rbac/AccessControlService.java

@@ -113,11 +113,11 @@ private boolean isAccessible(AuthenticatedUser user, AccessContext context) {
     return context.isAccessible(getUserPermissions(user, context.cluster()));
   }
 
-  private List<Permission> getUserPermissions(AuthenticatedUser user, String clusterName) {
+  private List<Permission> getUserPermissions(AuthenticatedUser user, @Nullable String clusterName) {
     return properties.getRoles()
         .stream()
         .filter(filterRole(user))
-        .filter(role -> role.getClusters().stream().anyMatch(clusterName::equalsIgnoreCase))
+        .filter(role -> clusterName == null || role.getClusters().stream().anyMatch(clusterName::equalsIgnoreCase))
         .flatMap(role -> role.getPermissions().stream())
         .toList();
   }

api/src/test/java/io/kafbat/ui/AbstractIntegrationTest.java

@@ -5,6 +5,7 @@
 import io.kafbat.ui.container.KafkaConnectContainer;
 import io.kafbat.ui.container.KsqlDbContainer;
 import io.kafbat.ui.container.SchemaRegistryContainer;
+import java.io.FileNotFoundException;
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Properties;
@@ -22,6 +23,7 @@
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.util.TestSocketUtils;
+import org.springframework.util.ResourceUtils;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.utility.DockerImageName;
@@ -75,6 +77,18 @@ public static class Initializer
     public void initialize(@NotNull ConfigurableApplicationContext context) {
       System.setProperty("kafka.clusters.0.name", LOCAL);
       System.setProperty("kafka.clusters.0.bootstrapServers", kafka.getBootstrapServers());
+
+      // Add ProtobufFileSerde configuration
+      System.setProperty("kafka.clusters.0.serde.0.name", "ProtobufFile");
+      System.setProperty("kafka.clusters.0.serde.0.topicValuesPattern", "masking-test-.*");
+      try {
+        System.setProperty("kafka.clusters.0.serde.0.properties.protobufFilesDir",
+            ResourceUtils.getFile("classpath:protobuf-serde").getAbsolutePath());
+      } catch (FileNotFoundException e) {
+        throw new RuntimeException(e);
+      }
+      System.setProperty("kafka.clusters.0.serde.0.properties.protobufMessageName", "test.MessageWithAny");
+
       // List unavailable hosts to verify failover
       System.setProperty("kafka.clusters.0.schemaRegistry",
           String.format("http://localhost:%1$s,http://localhost:%1$s,%2$s",