Merge branch 'main' into issues/688

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

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);
   }
 
@@ -243,11 +242,23 @@ public static class NgramProperties {
   @NoArgsConstructor
   @AllArgsConstructor
   public static class ClusterFtsProperties {
-    boolean enabled = false;
+    boolean enabled = true;
+    boolean defaultEnabled = false;
     NgramProperties schemas = new NgramProperties(1, 4);
     NgramProperties consumers = new NgramProperties(1, 4);
     NgramProperties connect = new NgramProperties(1, 4);
     NgramProperties acl = new NgramProperties(1, 4);
+
+    public boolean use(Boolean request) {
+      if (enabled) {
+        if (Boolean.TRUE.equals(request)) {
+          return true;
+        } else if (request == null && defaultEnabled) {
+          return true;
+        }
+      }
+      return false;
+    }
   }
 
   @PostConstruct

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

@@ -25,7 +25,7 @@
 
 @RestController
 @RequiredArgsConstructor
-public class AclsController extends AbstractController implements AclsApi, McpTool {
+public  class AclsController extends AbstractController implements AclsApi, McpTool {
 
   private final AclsService aclsService;
 
@@ -69,6 +69,7 @@ public Mono<ResponseEntity<Flux<KafkaAclDTO>>> listAcls(String clusterName,
                                                           String resourceName,
                                                           KafkaAclNamePatternTypeDTO namePatternTypeDto,
                                                           String search,
+                                                          Boolean fts,
                                                           ServerWebExchange exchange) {
     AccessContext context = AccessContext.builder()
         .cluster(clusterName)
@@ -89,7 +90,7 @@ public Mono<ResponseEntity<Flux<KafkaAclDTO>>> listAcls(String clusterName,
     return validateAccess(context).then(
         Mono.just(
             ResponseEntity.ok(
-                aclsService.listAcls(getCluster(clusterName), filter, search)
+                aclsService.listAcls(getCluster(clusterName), filter, search, fts)
                     .map(ClusterMapper::toKafkaAclDto)))
     ).doOnEach(sig -> audit(context, sig));
   }

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

@@ -128,6 +128,7 @@ public Mono<ResponseEntity<ConsumerGroupsPageResponseDTO>> getConsumerGroupsPage
       String search,
       ConsumerGroupOrderingDTO orderBy,
       SortOrderDTO sortOrderDto,
+      Boolean fts,
       ServerWebExchange exchange) {
 
     var context = AccessContext.builder()

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
@@ -129,6 +133,7 @@ public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
       String search,
       ConnectorColumnsToSortDTO orderBy,
       SortOrderDTO sortOrder,
+      Boolean fts,
       ServerWebExchange exchange
   ) {
     var context = AccessContext.builder()
@@ -140,7 +145,7 @@ public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
         ? getConnectorsComparator(orderBy)
         : getConnectorsComparator(orderBy).reversed();
 
-    Flux<FullConnectorInfoDTO> job = kafkaConnectService.getAllConnectors(getCluster(clusterName), search)
+    Flux<FullConnectorInfoDTO> job = kafkaConnectService.getAllConnectors(getCluster(clusterName), search, fts)
         .filterWhen(dto -> accessControlService.isConnectAccessible(dto.getConnect(), clusterName))
         .sort(comparator);
 

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

@@ -217,13 +217,15 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
                                                                     @Valid String search,
                                                                     SchemaColumnsToSortDTO orderBy,
                                                                     SortOrderDTO sortOrder,
+                                                                    Boolean fts,
                                                                     ServerWebExchange serverWebExchange) {
     var context = AccessContext.builder()
         .cluster(clusterName)
         .operationName("getSchemas")
         .build();
 
-    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
+    ClustersProperties.ClusterFtsProperties ftsProperties = clustersProperties.getFts();
+    boolean useFts = ftsProperties.use(fts);
 
     return schemaRegistryService
         .getAllSubjectNames(getCluster(clusterName))
@@ -234,7 +236,7 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
           int pageSize = perPage != null && perPage > 0 ? perPage : DEFAULT_PAGE_SIZE;
           int subjectToSkip = ((pageNum != null && pageNum > 0 ? pageNum : 1) - 1) * pageSize;
 
-          SchemasFilter filter = new SchemasFilter(subjects, fts.isEnabled(), fts.getSchemas());
+          SchemasFilter filter = new SchemasFilter(subjects, useFts, ftsProperties.getSchemas());
           List<String> filteredSubjects = new ArrayList<>(filter.find(search));
 
           var totalPages = (filteredSubjects.size() / pageSize)

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

@@ -174,20 +174,22 @@ public Mono<ResponseEntity<TopicsResponseDTO>> getTopics(String clusterName,
                                                            @Valid String search,
                                                            @Valid TopicColumnsToSortDTO orderBy,
                                                            @Valid SortOrderDTO sortOrder,
+                                                           Boolean fts,
                                                            ServerWebExchange exchange) {
 
     AccessContext context = AccessContext.builder()
         .cluster(clusterName)
         .operationName("getTopics")
         .build();
 
-    return topicsService.getTopicsForPagination(getCluster(clusterName), search, showInternal)
+    return topicsService.getTopicsForPagination(getCluster(clusterName), search, showInternal, fts)
         .flatMap(topics -> accessControlService.filterViewableTopics(topics, clusterName))
         .flatMap(topics -> {
           int pageSize = perPage != null && perPage > 0 ? perPage : DEFAULT_PAGE_SIZE;
           var topicsToSkip = ((page != null && page > 0 ? page : 1) - 1) * pageSize;
-          ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
-          Comparator<InternalTopic> comparatorForTopic = getComparatorForTopic(orderBy, fts.isEnabled());
+          ClustersProperties.ClusterFtsProperties ftsProperties = clustersProperties.getFts();
+          boolean useFts = ftsProperties.use(fts);
+          Comparator<InternalTopic> comparatorForTopic = getComparatorForTopic(orderBy, useFts);
           var comparator = sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
               ? comparatorForTopic : comparatorForTopic.reversed();