Merged main

Author: germanosin

.github/workflows/cve_checks.yml

@@ -68,7 +68,7 @@ jobs:
           cache-to: type=local,dest=/tmp/.buildx-cache
 
       - name: Run CVE checks
-        uses: aquasecurity/trivy-action@18f2510ee396bbf400402947b394f2dd8c87dbb0 # infered from @v0.29.0
+        uses: aquasecurity/trivy-action@dc5a429b52fcf669ce959baa2c2dd26090d2a6c4 # infered from @v0.32.0
         with:
           image-ref: "ghcr.io/kafbat/kafka-ui:latest"
           format: "table"

README.md

@@ -43,17 +43,23 @@ We extend our gratitude to Provectus for their past support in groundbreaking wo
 ![Interface](https://raw.githubusercontent.com/kafbat/kafka-ui/images/overview.gif)
 
 # Features
-* **Multi-Cluster Management** — monitor and manage all your clusters in one place
-* **Performance Monitoring with Metrics Dashboard** —  track key Kafka metrics with a lightweight dashboard
-* **View Kafka Brokers** — view topic and partition assignments, controller status
-* **View Kafka Topics** — view partition count, replication status, and custom configuration
-* **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 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
+
+* **Topic Insights** – View essential topic details including partition count, replication status, and custom configurations.
+* **Configuration Wizard** – Set up and configure your Kafka clusters directly through the UI.
+* **Multi-Cluster Management** – Monitor and manage all your Kafka clusters in one unified interface.
+* **Metrics Dashboard** – Track key Kafka metrics in real time with a streamlined, lightweight dashboard.
+* **Kafka Brokers Overview** – Inspect brokers, including partition assignments and controller status.
+* **Consumer Group Details** – Analyze parked offsets per partition, and monitor both combined and partition-specific lag.
+* **Message Browser** – Explore messages in JSON, plain text, or Avro encoding formats. Live view is supported, enriched with user-defined CEL message filters.
+* **Dynamic Topic Management** – Create and configure new topics with flexible, real-time settings.
+* **Pluggable Authentication** – Secure your UI using OAuth 2.0 (GitHub, GitLab, Google), LDAP, or basic authentication.
+* **Cloud IAM Support** – Integrate with **GCP IAM**, **Azure IAM**, and **AWS IAM** for cloud-native identity and access management.
+* **Managed Kafka Service Support** – Full support for **Azure EventHub**, **Google Cloud Managed Service for Apache Kafka**, and **AWS Managed Streaming for Apache Kafka (MSK)**—both server-based and serverless.
+* **Custom SerDe Plugin Support** – Use built-in serializers/deserializers like AWS Glue and Smile, or create your own custom plugins.
+* **Role-Based Access Control** – [Manage granular UI permissions](https://ui.docs.kafbat.io/configuration/rbac-role-based-access-control) with RBAC.
+* **Data Masking** – [Obfuscate sensitive data](https://ui.docs.kafbat.io/configuration/data-masking) in topic messages to enhance privacy and compliance.
+* **MCP Server** - [Model Context Protocol](https://ui.docs.kafbat.io/faq/mcp) Server
+
 
 ## Feature overview
 

api/Dockerfile

@@ -1,7 +1,7 @@
 # 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.6-jre-headless@sha256:75c5cc1ca1429513b56e9cbe3121bce86476cdec18b5b74b6842ab0af4b5a57f
+FROM azul/zulu-openjdk-alpine:21.0.8-jre-headless@sha256:9c7b4b7850bd4cdd78f91b369accc5b55beffa9a073b9a2bb94caa42606b9444
 
 RUN apk add --no-cache \
     # snappy codec

api/build.gradle

@@ -25,6 +25,7 @@ dependencies {
     implementation libs.spring.starter.oauth2.client
     implementation libs.spring.security.oauth2.resource.server
     implementation libs.spring.boot.actuator
+
     compileOnly libs.spring.boot.devtools
 
     implementation libs.spring.security.ldap
@@ -49,6 +50,7 @@ dependencies {
 
     implementation libs.jackson.databind.nullable
     implementation libs.cel
+    implementation libs.caffeine
     antlr libs.antlr
     implementation libs.antlr.runtime
 

api/src/main/java/io/kafbat/ui/config/ClustersProperties.java

@@ -6,6 +6,7 @@
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -39,6 +40,8 @@ public class ClustersProperties {
 
   MetricsStorage defaultMetricsStorage = new MetricsStorage();
 
+  CacheProperties cache = new CacheProperties();
+
   @Data
   public static class Cluster {
     @NotBlank(message = "field name for for cluster could not be blank")
@@ -205,6 +208,15 @@ public enum LogLevel {
     }
   }
 
+  @Data
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class CacheProperties {
+    boolean enabled = true;
+    Duration connectCacheExpiry = Duration.ofMinutes(1);
+    Duration connectClusterCacheExpiry = Duration.ofHours(24);
+  }
+
   @PostConstruct
   public void validateAndSetDefaults() {
     if (clusters != null) {

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

@@ -45,10 +45,12 @@ public class KafkaConnectController extends AbstractController implements KafkaC
 
   @Override
   public Mono<ResponseEntity<Flux<ConnectDTO>>> getConnects(String clusterName,
+                                                            Boolean withStats,
                                                             ServerWebExchange exchange) {
 
-    Flux<ConnectDTO> availableConnects = kafkaConnectService.getConnects(getCluster(clusterName))
-        .filterWhen(dto -> accessControlService.isConnectAccessible(dto, clusterName));
+    Flux<ConnectDTO> availableConnects = kafkaConnectService.getConnects(
+        getCluster(clusterName), withStats != null ? withStats : false
+        ).filterWhen(dto -> accessControlService.isConnectAccessible(dto, clusterName));
 
     return Mono.just(ResponseEntity.ok(availableConnects));
   }

api/src/main/java/io/kafbat/ui/mapper/KafkaConnectMapper.java

@@ -1,18 +1,23 @@
 package io.kafbat.ui.mapper;
 
+import io.kafbat.ui.config.ClustersProperties;
+import io.kafbat.ui.connect.model.ClusterInfo;
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
 import io.kafbat.ui.connect.model.ConnectorTask;
 import io.kafbat.ui.connect.model.NewConnector;
+import io.kafbat.ui.model.ConnectDTO;
 import io.kafbat.ui.model.ConnectorDTO;
 import io.kafbat.ui.model.ConnectorPluginConfigValidationResponseDTO;
 import io.kafbat.ui.model.ConnectorPluginDTO;
+import io.kafbat.ui.model.ConnectorStateDTO;
 import io.kafbat.ui.model.ConnectorStatusDTO;
 import io.kafbat.ui.model.ConnectorTaskStatusDTO;
 import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.TaskStatusDTO;
-import io.kafbat.ui.model.connect.InternalConnectInfo;
+import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import java.util.List;
+import java.util.Optional;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 
@@ -38,7 +43,60 @@ ConnectorPluginConfigValidationResponseDTO fromClient(
       io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse
           connectorPluginConfigValidationResponse);
 
-  default FullConnectorInfoDTO fullConnectorInfo(InternalConnectInfo connectInfo) {
+  default ConnectDTO toKafkaConnect(
+      ClustersProperties.ConnectCluster connect,
+      List<InternalConnectorInfo> connectors,
+      ClusterInfo clusterInfo,
+      boolean withStats) {
+    Integer connectorCount = null;
+    Integer failedConnectors = null;
+    Integer tasksCount = null;
+    Integer failedTasksCount = null;
+
+    if (withStats) {
+      connectorCount = connectors.size();
+      failedConnectors = 0;
+      tasksCount = 0;
+      failedTasksCount = 0;
+
+      for (InternalConnectorInfo connector : connectors) {
+        Optional<ConnectorDTO> internalConnector = Optional.ofNullable(connector.getConnector());
+
+        failedConnectors += internalConnector
+            .map(ConnectorDTO::getStatus)
+            .map(ConnectorStatusDTO::getState)
+            .filter(ConnectorStateDTO.FAILED::equals)
+            .map(s -> 1).orElse(0);
+
+        tasksCount += internalConnector.map(ConnectorDTO::getTasks).map(List::size).orElse(0);
+
+        if (connector.getTasks() != null) {
+          failedTasksCount += (int) connector.getTasks().stream()
+              .filter(t ->
+                  Optional.ofNullable(t)
+                      .map(TaskDTO::getStatus)
+                      .map(TaskStatusDTO::getState)
+                      .map(ConnectorTaskStatusDTO.FAILED::equals)
+                      .orElse(false)
+              ).count();
+        }
+      }
+
+    }
+
+    return new ConnectDTO()
+        .address(connect.getAddress())
+        .name(connect.getName())
+        .connectorsCount(connectorCount)
+        .failedConnectorsCount(failedConnectors)
+        .tasksCount(tasksCount)
+        .failedTasksCount(failedTasksCount)
+        .version(clusterInfo.getVersion())
+        .commit(clusterInfo.getCommit())
+        .clusterId(clusterInfo.getKafkaClusterId());
+  }
+
+  default FullConnectorInfoDTO fullConnectorInfo(InternalConnectorInfo connectInfo) {
     ConnectorDTO connector = connectInfo.getConnector();
     List<TaskDTO> tasks = connectInfo.getTasks();
     int failedTasksCount = (int) tasks.stream()

api/src/main/java/io/kafbat/ui/model/connect/InternalConnectInfo.java renamed to api/src/main/java/io/kafbat/ui/model/connect/InternalConnectorInfo.java

@@ -9,7 +9,7 @@
 
 @Data
 @Builder(toBuilder = true)
-public class InternalConnectInfo {
+public class InternalConnectorInfo {
   private final ConnectorDTO connector;
   private final Map<String, Object> config;
   private final List<TaskDTO> tasks;

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

@@ -112,7 +112,7 @@ void configure(Configuration configuration) {
         && configuration.defaultKeyMessageDescriptor() == null
         && configuration.messageDescriptorMap().isEmpty()
         && configuration.keyMessageDescriptorMap().isEmpty()) {
-      throw new ValidationException("Neither default, not per-topic descriptors defined for " + name() + " serde");
+      throw new ValidationException("Neither default, nor per-topic descriptors defined for " + name() + " serde");
     }
     this.defaultMessageDescriptor = configuration.defaultMessageDescriptor();
     this.defaultKeyMessageDescriptor = configuration.defaultKeyMessageDescriptor();