Merge branch 'main' into issues/1440

Author: Haarolean

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/client/RetryingKafkaConnectClient.java

@@ -7,11 +7,13 @@
 import io.kafbat.ui.connect.ApiClient;
 import io.kafbat.ui.connect.api.KafkaConnectClientApi;
 import io.kafbat.ui.connect.model.Connector;
+import io.kafbat.ui.connect.model.ConnectorExpand;
 import io.kafbat.ui.connect.model.ConnectorPlugin;
 import io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse;
 import io.kafbat.ui.connect.model.ConnectorStatus;
 import io.kafbat.ui.connect.model.ConnectorTask;
 import io.kafbat.ui.connect.model.ConnectorTopics;
+import io.kafbat.ui.connect.model.ExpandedConnector;
 import io.kafbat.ui.connect.model.NewConnector;
 import io.kafbat.ui.connect.model.TaskStatus;
 import io.kafbat.ui.exception.KafkaConnectConflictResponseException;
@@ -221,13 +223,17 @@ public Mono<ResponseEntity<Map<String, ConnectorTopics>>> getConnectorTopicsWith
   }
 
   @Override
-  public Mono<List<String>> getConnectors(String search) throws WebClientResponseException {
-    return withRetryOnConflictOrRebalance(super.getConnectors(search));
+  public Mono<Map<String, ExpandedConnector>> getConnectors(
+      String search, List<ConnectorExpand> expand
+  ) throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.getConnectors(search, expand));
   }
 
   @Override
-  public Mono<ResponseEntity<List<String>>> getConnectorsWithHttpInfo(String search) throws WebClientResponseException {
-    return withRetryOnConflictOrRebalance(super.getConnectorsWithHttpInfo(search));
+  public Mono<ResponseEntity<Map<String, ExpandedConnector>>> getConnectorsWithHttpInfo(
+      String search, List<ConnectorExpand> expand
+  ) throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.getConnectorsWithHttpInfo(search, expand));
   }
 
   @Override

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

@@ -227,7 +227,6 @@ public enum LogLevel {
   @AllArgsConstructor
   public static class CacheProperties {
     boolean enabled = true;
-    Duration connectCacheExpiry = Duration.ofMinutes(1);
     Duration connectClusterCacheExpiry = Duration.ofHours(24);
   }
 

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

@@ -66,8 +66,12 @@ public Mono<ResponseEntity<Flux<String>>> getConnectors(String clusterName, Stri
         .build();
 
     return validateAccess(context)
-        .thenReturn(ResponseEntity.ok(kafkaConnectService.getConnectorNames(getCluster(clusterName), connectName)))
-        .doOnEach(sig -> audit(context, sig));
+        .thenReturn(
+            ResponseEntity.ok(
+              kafkaConnectService.getConnectors(getCluster(clusterName), connectName)
+                  .flatMapMany(m -> Flux.fromIterable(m.keySet()))
+            )
+        ).doOnEach(sig -> audit(context, sig));
   }
 
   @Override

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

@@ -2,8 +2,11 @@
 
 import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.connect.model.ClusterInfo;
+import io.kafbat.ui.connect.model.Connector;
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
 import io.kafbat.ui.connect.model.ConnectorTask;
+import io.kafbat.ui.connect.model.ConnectorTopics;
+import io.kafbat.ui.connect.model.ExpandedConnector;
 import io.kafbat.ui.connect.model.NewConnector;
 import io.kafbat.ui.model.ConnectDTO;
 import io.kafbat.ui.model.ConnectorDTO;
@@ -14,21 +17,41 @@
 import io.kafbat.ui.model.ConnectorTaskStatusDTO;
 import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.TaskDTO;
+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;
 import java.util.Optional;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 
 @Mapper(componentModel = "spring")
 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);
 
+  default ConnectorDTO fromClient(Connector connector, ConnectorTopics topics) {
+    ConnectorDTO connectorDto = this.fromClient(connector);
+    if (topics != null) {
+      return connectorDto.topics(topics.getTopics());
+    }
+    return connectorDto;
+  }
+
   ConnectorStatusDTO fromClient(ConnectorStatusConnector connectorStatus);
 
   @Mapping(target = "status", ignore = true)
@@ -43,6 +66,39 @@ ConnectorPluginConfigValidationResponseDTO fromClient(
       io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse
           connectorPluginConfigValidationResponse);
 
+  default InternalConnectorInfo fromClient(String connect, ExpandedConnector connector, @Nullable List<String> topics) {
+    Objects.requireNonNull(connector.getInfo());
+    Objects.requireNonNull(connector.getStatus());
+    List<TaskDTO> tasks = List.of();
+
+    if (connector.getInfo().getTasks() != null
+        && connector.getStatus().getTasks() != null
+    ) {
+      Map<Integer, TaskIdDTO> taskIds = connector.getInfo().getTasks()
+          .stream().map(t -> new TaskIdDTO().task(t.getTask()).connector(t.getConnector()))
+          .collect(Collectors.toMap(
+              TaskIdDTO::getTask,
+              t -> t
+          ));
+
+      tasks = connector.getStatus().getTasks().stream()
+          .map(s ->
+              new TaskDTO().status(fromClient(s)).id(taskIds.get(s.getId()))
+          ).toList();
+    }
+
+    ConnectorDTO connectorDto = fromClient(connector.getInfo())
+        .connect(connect)
+        .status(fromClient(connector.getStatus().getConnector()));
+
+    return InternalConnectorInfo.builder()
+        .connector(connectorDto)
+        .config(connector.getInfo().getConfig())
+        .tasks(tasks)
+        .topics(topics)
+        .build();
+  }
+
   default ConnectDTO toKafkaConnect(
       ClustersProperties.ConnectCluster connect,
       List<InternalConnectorInfo> connectors,
@@ -114,4 +170,21 @@ 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/KafkaConfigSanitizer.java

@@ -19,7 +19,7 @@
 import org.springframework.stereotype.Component;
 
 @Component
-class KafkaConfigSanitizer {
+public class KafkaConfigSanitizer {
 
   private static final String SANITIZED_VALUE = "******";