Merge branch 'main' into DINF-2950-schema-prefix-fix

Author: ashok-pnt

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:48356cf1b81243b6e236509bb6ae281969fc0eade03f622220973a4e224cd768
 
 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);
   }
 
@@ -237,6 +236,7 @@ public static class CacheProperties {
   public static class NgramProperties {
     int ngramMin = 1;
     int ngramMax = 4;
+    boolean distanceScore = true;
   }
 
   @Data
@@ -245,10 +245,10 @@ public static class NgramProperties {
   public static class ClusterFtsProperties {
     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);
+    NgramProperties schemas = new NgramProperties(1, 4, true);
+    NgramProperties consumers = new NgramProperties(1, 4, true);
+    NgramProperties connect = new NgramProperties(1, 4, true);
+    NgramProperties acl = new NgramProperties(1, 4, true);
 
     public boolean use(Boolean request) {
       if (enabled) {

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

@@ -23,6 +23,7 @@
 import io.kafbat.ui.service.mcp.McpTool;
 import java.util.Comparator;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import javax.validation.Valid;
 import lombok.RequiredArgsConstructor;
@@ -66,8 +67,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
@@ -137,15 +142,18 @@ public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
         .operationName("getAllConnectors")
         .build();
 
+    var maybeComparator = Optional.ofNullable(orderBy).map(this::getConnectorsComparator);
+
     var comparator = sortOrder == null || sortOrder.equals(SortOrderDTO.ASC)
-        ? getConnectorsComparator(orderBy)
-        : getConnectorsComparator(orderBy).reversed();
+        ? maybeComparator
+        : maybeComparator.map(Comparator::reversed);
 
-    Flux<FullConnectorInfoDTO> job = kafkaConnectService.getAllConnectors(getCluster(clusterName), search, fts)
-        .filterWhen(dto -> accessControlService.isConnectAccessible(dto.getConnect(), clusterName))
-        .sort(comparator);
+    Flux<FullConnectorInfoDTO> connectors = kafkaConnectService.getAllConnectors(getCluster(clusterName), search, fts)
+        .filterWhen(dto -> accessControlService.isConnectAccessible(dto.getConnect(), clusterName));
 
-    return Mono.just(ResponseEntity.ok(job))
+    Flux<FullConnectorInfoDTO> sorted = comparator.map(connectors::sort).orElse(connectors);
+
+    return Mono.just(ResponseEntity.ok(sorted))
         .doOnEach(sig -> audit(context, sig));
   }
 
@@ -280,9 +288,7 @@ private Comparator<FullConnectorInfoDTO> getConnectorsComparator(ConnectorColumn
         FullConnectorInfoDTO::getName,
         Comparator.nullsFirst(Comparator.naturalOrder())
     );
-    if (orderBy == null) {
-      return defaultComparator;
-    }
+
     return switch (orderBy) {
       case CONNECT -> Comparator.comparing(
           FullConnectorInfoDTO::getConnect,

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

@@ -25,6 +25,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import javax.validation.Valid;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -244,11 +245,15 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
 
           List<String> subjectsToRetrieve;
           boolean paginate = true;
-          var schemaComparator = getComparatorForSchema(orderBy);
-          final Comparator<SubjectWithCompatibilityLevel> comparator =
-              sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
-                  ? schemaComparator : schemaComparator.reversed();
+
+          var schemaComparator = Optional.ofNullable(orderBy).map(this::getComparatorForSchema);
+          var comparator = sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
+              ? schemaComparator : schemaComparator.map(Comparator::reversed);
+
           if (orderBy == null || SchemaColumnsToSortDTO.SUBJECT.equals(orderBy)) {
+            if (orderBy != null) {
+              filteredSubjects.sort(Comparator.nullsFirst(Comparator.naturalOrder()));
+            }
             if (SortOrderDTO.DESC.equals(sortOrder)) {
               filteredSubjects.sort(Comparator.nullsFirst(Comparator.reverseOrder()));
             }
@@ -274,11 +279,13 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
 
   private List<SubjectWithCompatibilityLevel> paginateSchemas(
       List<SubjectWithCompatibilityLevel> subjects,
-      Comparator<SubjectWithCompatibilityLevel> comparator,
+      Optional<Comparator<SubjectWithCompatibilityLevel>> comparator,
       boolean paginate,
       int pageSize,
       int subjectToSkip) {
-    subjects.sort(comparator);
+
+    comparator.ifPresent(subjects::sort);
+
     if (paginate) {
       return subjects.subList(subjectToSkip, Math.min(subjectToSkip + pageSize, subjects.size()));
     } else {

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