Merge branch 'main' into auth_page_be

.github/dependabot.yml

@@ -14,6 +14,23 @@ updates:
     - "type/dependencies"
     - "scope/backend"
 
+- package-ecosystem: docker
+  directory: "/api"
+  schedule:
+    interval: weekly
+    time: "10:00"
+    timezone: Europe/London
+  reviewers:
+    - "kafbat/backend"
+  open-pull-requests-limit: 10
+  ignore:
+  - dependency-name: "azul/zulu-openjdk-alpine"
+    # Limit dependabot pull requests to minor Java upgrades
+    update-types: ["version-update:semver-major"]
+  labels:
+    - "type/dependencies"
+    - "scope/backend"
+
 - package-ecosystem: npm
   directory: "/frontend"
   schedule:

.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>  •
@@ -50,7 +50,7 @@ We extend our gratitude to Provectus for their past support in groundbreaking wo
 * **View Consumer Groups** — view per-partition parked offsets, combined and per-partition lag
 * **Browse Messages** — browse messages with JSON, plain text, and Avro encoding
 * **Dynamic Topic Configuration** — create and configure new topics with dynamic configuration
-* **Configurable Authentification** — [secure](https://ui.docs.kafbat.io/configuration/authentication) your installation with optional Github/Gitlab/Google OAuth 2.0
+* **Configurable Authentication** — [secure](https://ui.docs.kafbat.io/configuration/authentication) your installation with optional Github/Gitlab/Google OAuth 2.0
 * **Custom serialization/deserialization plugins** - [use](https://ui.docs.kafbat.io/configuration/serialization-serde) a ready-to-go serde for your data like AWS Glue or Smile, or code your own!
 * **Role based access control** - [manage permissions](https://ui.docs.kafbat.io/configuration/rbac-role-based-access-control) to access the UI with granular precision
 * **Data masking** - [obfuscate](https://ui.docs.kafbat.io/configuration/data-masking) sensitive data in topic messages

api/Dockerfile

@@ -1,4 +1,7 @@
-FROM azul/zulu-openjdk-alpine:17.0.11-jre-headless
+# 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: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/config/auth/OAuthSecurityConfig.java

@@ -8,7 +8,7 @@
 import java.util.Map;
 import java.util.Optional;
 import lombok.RequiredArgsConstructor;
-import lombok.extern.log4j.Log4j2;
+import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.Nullable;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.security.oauth2.client.OAuth2ClientProperties;
@@ -41,7 +41,7 @@
 @EnableWebFluxSecurity
 @EnableReactiveMethodSecurity
 @RequiredArgsConstructor
-@Log4j2
+@Slf4j
 public class OAuthSecurityConfig extends AbstractAuthSecurityConfig {
 
   private final OAuthProperties properties;

api/src/main/java/io/kafbat/ui/controller/MessagesController.java

@@ -118,10 +118,11 @@ public Mono<ResponseEntity<Flux<TopicMessageEventDTO>>> getTopicMessagesV2(Strin
     if (cursor != null) {
       messagesFlux = messagesService.loadMessages(getCluster(clusterName), topicName, cursor);
     } else {
+      var pollingMode = mode == null ? PollingModeDTO.LATEST : mode;
       messagesFlux = messagesService.loadMessages(
           getCluster(clusterName),
           topicName,
-          ConsumerPosition.create(checkNotNull(mode), checkNotNull(topicName), partitions, timestamp, offset),
+          ConsumerPosition.create(pollingMode, checkNotNull(topicName), partitions, timestamp, offset),
           stringFilter,
           smartFilterId,
           limit,

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/ProducerRecordCreator.java

@@ -31,7 +31,7 @@ public ProducerRecord<byte[], byte[]> create(String topic,
 
   private Iterable<Header> createHeaders(Map<String, String> clientHeaders) {
     RecordHeaders headers = new RecordHeaders();
-    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
+    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v == null ? null : v.getBytes())));
     return headers;
   }
 

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();
   }