Merge branch 'main' into issues/1333-optional-fts-fe

Author: Vixtir

README.md

@@ -72,28 +72,25 @@ Kafbat UI wraps major functions of Apache Kafka with an intuitive user interface
 ![Interface](documentation/images/Interface.gif)
 
 ## Topics
-Kafbat UI makes it easy for you to create topics in your browser by several clicks,
-pasting your own parameters, and viewing topics in the list.
+Kafbat UI makes it easy for you to create topics in your browser with just a few clicks, by pasting your own parameters, and viewing topics in the list.
 
 ![Create Topic](documentation/images/Create_topic_kafka-ui.gif)
 
-It's possible to jump from connectors view to corresponding topics and from a topic to consumers (back and forth) for more convenient navigation.
-connectors, overview topic settings.
+You can jump from the connectors view to corresponding topics and from a topic to consumers (back and forth) for more convenient navigation, including connectors and overview topic settings.
 
 ![Connector_Topic_Consumer](documentation/images/Connector_Topic_Consumer.gif)
 
 ### Messages
-Let's say we want to produce messages for our topic. With the Kafbat UI we can send or write data/messages to the Kafka topics without effort by specifying parameters, and viewing messages in the list.
+Suppose you want to produce messages for your topic. With Kafbat UI, you can easily send or write data/messages to Kafka topics by specifying parameters and viewing messages in the list.
 
 ![Produce Message](documentation/images/Create_message_kafka-ui.gif)
 
 ## Schema registry
-There are 3 supported types of schemas: Avro®, JSON Schema, and Protobuf schemas.
+There are three supported types of schemas: Avro®, JSON Schema, and Protobuf schemas.
 
 ![Create Schema Registry](documentation/images/Create_schema.gif)
 
-Before producing avro/protobuf encoded messages, you have to add a schema for the topic in Schema Registry. Now all these steps are easy to do
-with a few clicks in a user-friendly interface.
+Before producing Avro/Protobuf encoded messages, you need to add a schema for the topic in the Schema Registry. All these steps are now easy to do with just a few clicks in a user-friendly interface.
 
 ![Avro Schema Topic](documentation/images/Schema_Topic.gif)
 
@@ -111,7 +108,7 @@ docker run -it -p 8080:8080 -e DYNAMIC_CONFIG_ENABLED=true ghcr.io/kafbat/kafka-
 
 Then access the web UI at [http://localhost:8080](http://localhost:8080)
 
-The command is sufficient to try things out. When you're done trying things out, you can proceed with a [persistent installation](https://ui.docs.kafbat.io/quick-start/persistent-start)
+This command is sufficient to try things out. When you're done, you can proceed with a [persistent installation](https://ui.docs.kafbat.io/quick-start/persistent-start).
 
 ## Persistent installation
 
@@ -146,24 +143,24 @@ Please refer to our [configuration](https://ui.docs.kafbat.io/configuration/conf
 
 ## Building from sources
 
-[Quick start](https://ui.docs.kafbat.io/development/building/prerequisites) with building
+[Quick start](https://ui.docs.kafbat.io/development/building/prerequisites) for building from source
 
 ## Liveliness and readiness probes
-Liveliness and readiness endpoint is at `/actuator/health`.<br/>
-Info endpoint (build info) is located at `/actuator/info`.
+The liveness and readiness endpoint is at `/actuator/health`.<br/>
+The info endpoint (build info) is located at `/actuator/info`.
 
 # Configuration options
 
-All the environment variables/config properties could be found [here](https://ui.docs.kafbat.io/configuration/misc-configuration-properties).
+All environment variables and configuration properties can be found [here](https://ui.docs.kafbat.io/configuration/misc-configuration-properties).
 
 # Contributing
 
-Please refer to [contributing guide](https://ui.docs.kafbat.io/development/contributing), we'll guide you from there.
+Please refer to the [contributing guide](https://ui.docs.kafbat.io/development/contributing); we'll guide you from there.
 
 # Support
 
 As we're fully independent, team members contribute in their free time.
-Your support is crucial for us, if you wish to sponsor us, take a look [here](https://github.com/sponsors/kafbat) 
+Your support is crucial for us, if you wish to sponsor us, take a look [here](https://github.com/sponsors/kafbat)
 
 # Powered by
 

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

@@ -44,12 +44,12 @@ public McpAsyncServer mcpServer(WebFluxSseServerTransportProvider transport) {
     // Configure server capabilities with resource support
     var capabilities = McpSchema.ServerCapabilities.builder()
         .resources(false, true)
-        .tools(true) // Tool support with list changes notifications
+        .tools(true) // Tools support with list changes notifications
         .prompts(false) // Prompt support with list changes notifications
         .logging() // Logging support
         .build();
 
-    // Create the server with both tool and resource capabilities
+    // Create the server with both tools and resource capabilities
     return McpServer.async(transport)
         .serverInfo("Kafka UI MCP", "0.0.1")
         .capabilities(capabilities)

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

@@ -11,6 +11,7 @@
 import io.kafbat.ui.api.TopicsApi;
 import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.mapper.ClusterMapper;
+import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.InternalTopic;
 import io.kafbat.ui.model.InternalTopicConfig;
 import io.kafbat.ui.model.PartitionsIncreaseDTO;
@@ -28,6 +29,8 @@
 import io.kafbat.ui.model.TopicUpdateDTO;
 import io.kafbat.ui.model.TopicsResponseDTO;
 import io.kafbat.ui.model.rbac.AccessContext;
+import io.kafbat.ui.model.rbac.permission.ConnectAction;
+import io.kafbat.ui.service.KafkaConnectService;
 import io.kafbat.ui.service.TopicsService;
 import io.kafbat.ui.service.analyze.TopicAnalysisService;
 import io.kafbat.ui.service.mcp.McpTool;
@@ -55,6 +58,7 @@ public class TopicsController extends AbstractController implements TopicsApi, M
   private final TopicAnalysisService topicAnalysisService;
   private final ClusterMapper clusterMapper;
   private final ClustersProperties clustersProperties;
+  private final KafkaConnectService kafkaConnectService;
 
   @Override
   public Mono<ResponseEntity<TopicDTO>> createTopic(
@@ -370,4 +374,23 @@ private Comparator<InternalTopic> getComparatorForTopic(
       default -> defaultComparator;
     };
   }
+
+  @Override
+  public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getTopicConnectors(String clusterName,
+                                                                             String topicName,
+                                                                             ServerWebExchange exchange) {
+    var context = AccessContext.builder()
+        .cluster(clusterName)
+        .topicActions(topicName, VIEW)
+        .operationName("getTopicConnectors")
+        .operationParams(topicName)
+        .build();
+
+    Flux<FullConnectorInfoDTO> job = kafkaConnectService.getTopicConnectors(getCluster(clusterName), topicName)
+        .filterWhen(dto -> accessControlService.isConnectAccessible(dto.getConnect(), clusterName));
+
+    return validateAccess(context)
+        .then(Mono.just(ResponseEntity.ok(job)))
+        .doOnEach(sig -> audit(context, sig));
+  }
 }

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

@@ -18,6 +18,7 @@
 import io.kafbat.ui.model.TaskIdDTO;
 import io.kafbat.ui.model.TaskStatusDTO;
 import io.kafbat.ui.model.connect.InternalConnectorInfo;
+import io.kafbat.ui.service.metrics.scrape.KafkaConnectState;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -31,6 +32,12 @@
 public interface KafkaConnectMapper {
   NewConnector toClient(io.kafbat.ui.model.NewConnectorDTO newConnector);
 
+  default ClusterInfo toClient(KafkaConnectState state) {
+    ClusterInfo clusterInfo = new ClusterInfo();
+    clusterInfo.setVersion(state.getVersion());
+    return clusterInfo;
+  }
+
   @Mapping(target = "status", ignore = true)
   @Mapping(target = "connect", ignore = true)
   ConnectorDTO fromClient(io.kafbat.ui.connect.model.Connector connector);
@@ -153,4 +160,23 @@ default FullConnectorInfoDTO fullConnectorInfo(InternalConnectorInfo connectInfo
         .tasksCount(tasks.size())
         .failedTasksCount(failedTasksCount);
   }
+
+  default KafkaConnectState toScrapeState(ConnectDTO connect, List<InternalConnectorInfo> connectors) {
+    return KafkaConnectState.builder()
+        .name(connect.getName())
+        .version(connect.getVersion().orElse("Unknown"))
+        .connectors(connectors.stream().map(this::toScrapeState).toList())
+        .build();
+  }
+
+  default KafkaConnectState.ConnectorState toScrapeState(InternalConnectorInfo connector) {
+    return new KafkaConnectState.ConnectorState(
+        connector.getConnector().getName(),
+        connector.getConnector().getType(),
+        connector.getConnector().getStatus(),
+        connector.getTopics()
+    );
+  }
+
+
 }

api/src/main/java/io/kafbat/ui/model/Statistics.java

@@ -1,8 +1,10 @@
 package io.kafbat.ui.model;
 
 import io.kafbat.ui.service.ReactiveAdminClient;
+import io.kafbat.ui.service.metrics.scrape.KafkaConnectState;
 import io.kafbat.ui.service.metrics.scrape.ScrapedClusterState;
 import java.util.List;
+import java.util.Map;
 import java.util.function.UnaryOperator;
 import java.util.stream.Stream;
 import lombok.Builder;
@@ -19,6 +21,7 @@ public class Statistics implements AutoCloseable {
   ReactiveAdminClient.ClusterDescription clusterDescription;
   Metrics metrics;
   ScrapedClusterState clusterState;
+  Map<String, KafkaConnectState> connectStates;
 
   public static Statistics empty() {
     return builder()
@@ -28,6 +31,7 @@ public static Statistics empty() {
         .clusterDescription(ReactiveAdminClient.ClusterDescription.empty())
         .metrics(Metrics.empty())
         .clusterState(ScrapedClusterState.empty())
+        .connectStates(Map.of())
         .build();
   }
 

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

@@ -1,7 +1,5 @@
 package io.kafbat.ui.service;
 
-import com.github.benmanes.caffeine.cache.AsyncCache;
-import com.github.benmanes.caffeine.cache.Caffeine;
 import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.connect.api.KafkaConnectClientApi;
 import io.kafbat.ui.connect.model.ClusterInfo;
@@ -25,12 +23,15 @@
 import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.KafkaCluster;
 import io.kafbat.ui.model.NewConnectorDTO;
+import io.kafbat.ui.model.Statistics;
 import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.TaskIdDTO;
 import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import io.kafbat.ui.service.index.KafkaConnectNgramFilter;
+import io.kafbat.ui.service.metrics.scrape.KafkaConnectState;
 import io.kafbat.ui.util.ReactiveFailover;
 import jakarta.validation.Valid;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -49,17 +50,16 @@ public class KafkaConnectService {
   private final KafkaConnectMapper kafkaConnectMapper;
   private final KafkaConfigSanitizer kafkaConfigSanitizer;
   private final ClustersProperties clustersProperties;
-  private final AsyncCache<String, ClusterInfo> cacheClusterInfo;
+  private final StatisticsCache statisticsCache;
 
   public KafkaConnectService(KafkaConnectMapper kafkaConnectMapper,
                              KafkaConfigSanitizer kafkaConfigSanitizer,
-                             ClustersProperties clustersProperties) {
+                             ClustersProperties clustersProperties,
+                             StatisticsCache statisticsCache) {
     this.kafkaConnectMapper = kafkaConnectMapper;
     this.kafkaConfigSanitizer = kafkaConfigSanitizer;
     this.clustersProperties = clustersProperties;
-    this.cacheClusterInfo = Caffeine.newBuilder()
-        .expireAfterWrite(clustersProperties.getCache().getConnectClusterCacheExpiry())
-        .buildAsync();
+    this.statisticsCache = statisticsCache;
   }
 
   public Flux<ConnectDTO> getConnects(KafkaCluster cluster, boolean withStats) {
@@ -89,14 +89,17 @@ public Flux<ConnectDTO> getConnects(KafkaCluster cluster, boolean withStats) {
     }
   }
 
-  private Mono<ClusterInfo> getClusterInfo(KafkaCluster cluster, String connectName) {
-    return Mono.fromFuture(cacheClusterInfo.get(connectName, (t, e) ->
-        api(cluster, connectName).mono(KafkaConnectClientApi::getClusterInfo)
-            .onErrorResume(th -> {
-              log.error("Error on collecting cluster info", th);
-              return Mono.just(new ClusterInfo());
-            }).toFuture()
-    ));
+  public Mono<ClusterInfo> getClusterInfo(KafkaCluster cluster, String connectName) {
+    KafkaConnectState state = statisticsCache.get(cluster).getConnectStates().get(connectName);
+    if (state != null) {
+      return Mono.just(kafkaConnectMapper.toClient(state));
+    } else {
+      return api(cluster, connectName).mono(KafkaConnectClientApi::getClusterInfo)
+          .onErrorResume(th -> {
+            log.error("Error on collecting cluster info", th);
+            return Mono.just(new ClusterInfo());
+          });
+    }
   }
 
   private Flux<InternalConnectorInfo> getConnectConnectors(
@@ -134,6 +137,33 @@ public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
         .flatMapMany(Flux::fromIterable);
   }
 
+  public Flux<KafkaConnectState> scrapeAllConnects(KafkaCluster cluster) {
+
+    Optional<List<ClustersProperties.@Valid ConnectCluster>> connectClusters =
+        Optional.ofNullable(cluster.getOriginalProperties().getKafkaConnect());
+
+    return Flux.fromIterable(connectClusters.orElse(List.of())).flatMap(c ->
+        getClusterInfo(cluster, c.getName()).map(info ->
+                kafkaConnectMapper.toKafkaConnect(c, List.of(), info, false)
+        ).onErrorResume((t) -> Mono.just(new ConnectDTO().name(c.getName())))
+    ).flatMap(connect ->
+        getConnectorsWithErrorsSuppress(cluster, connect.getName())
+            .onErrorResume(t -> Mono.just(Map.of()))
+            .flatMapMany(connectors ->
+                Flux.fromIterable(connectors.entrySet())
+                    .flatMap(e ->
+                        getConnectorTopics(
+                            cluster,
+                            connect.getName(),
+                            e.getKey()
+                        ).map(topics ->
+                            kafkaConnectMapper.fromClient(connect.getName(), e.getValue(), topics.getTopics())
+                        )
+                    )
+            ).collectList().map(connectors -> kafkaConnectMapper.toScrapeState(connect, connectors))
+    );
+  }
+
   private List<FullConnectorInfoDTO> filterConnectors(
       List<FullConnectorInfoDTO> connectors,
       String search,
@@ -349,4 +379,30 @@ public Mono<Void> resetConnectorOffsets(KafkaCluster cluster, String connectName
                       .formatted(connectorName, connectName));
             });
   }
+
+  public Flux<FullConnectorInfoDTO> getTopicConnectors(KafkaCluster cluster, String topicName) {
+    Map<String, KafkaConnectState> connectStates = this.statisticsCache.get(cluster).getConnectStates();
+    Map<String, List<String>> filteredConnects = new HashMap<>();
+    for (Map.Entry<String, KafkaConnectState> entry : connectStates.entrySet()) {
+      List<KafkaConnectState.ConnectorState> connectors =
+          entry.getValue().getConnectors().stream().filter(c -> c.topics().contains(topicName)).toList();
+      if (!connectors.isEmpty()) {
+        filteredConnects.put(entry.getKey(), connectors.stream().map(KafkaConnectState.ConnectorState::name).toList());
+      }
+    }
+
+    return Flux.fromIterable(filteredConnects.entrySet())
+        .flatMap(entry ->
+            getConnectorsWithErrorsSuppress(cluster, entry.getKey())
+                .map(connectors ->
+                        connectors.entrySet()
+                            .stream()
+                            .filter(c -> entry.getValue().contains(c.getKey()))
+                            .map(c -> kafkaConnectMapper.fromClient(entry.getKey(), c.getValue(), null))
+                            .map(kafkaConnectMapper::fullConnectorInfo)
+                            .toList()
+                )
+        ).flatMap(Flux::fromIterable);
+
+  }
 }