Merge branch 'main' into main

Author: survivant

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/config/ClustersProperties.java

@@ -243,11 +243,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/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

@@ -129,6 +129,7 @@ public Mono<ResponseEntity<Flux<FullConnectorInfoDTO>>> getAllConnectors(
       String search,
       ConnectorColumnsToSortDTO orderBy,
       SortOrderDTO sortOrder,
+      Boolean fts,
       ServerWebExchange exchange
   ) {
     var context = AccessContext.builder()
@@ -140,7 +141,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();
 

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

@@ -8,5 +8,7 @@ public enum ClusterFeature {
   KAFKA_ACL_VIEW,
   KAFKA_ACL_EDIT,
   CLIENT_QUOTA_MANAGEMENT,
-  GRAPHS_ENABLED
+  GRAPHS_ENABLED,
+  FTS_ENABLED,
+  FTS_DEFAULT_ENABLED
 }

api/src/main/java/io/kafbat/ui/serdes/BuiltInSerde.java

@@ -1,7 +1,9 @@
 package io.kafbat.ui.serdes;
 
 import io.kafbat.ui.serde.api.PropertyResolver;
+import io.kafbat.ui.serde.api.SchemaDescription;
 import io.kafbat.ui.serde.api.Serde;
+import java.util.Optional;
 
 public interface BuiltInSerde extends Serde {
 
@@ -24,4 +26,25 @@ default void configure(PropertyResolver serdeProperties,
                          PropertyResolver kafkaClusterProperties,
                          PropertyResolver globalProperties) {
   }
+
+  @Override
+  default boolean canSerialize(String topic, Serde.Target type) {
+    return false;
+  }
+
+  @Override
+  default Serde.Serializer serializer(String topic, Serde.Target type) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  default Optional<SchemaDescription> getSchema(String topic, Serde.Target type) {
+    return Optional.empty();
+  }
+
+  @Override
+  default Optional<String> getDescription() {
+    return Optional.empty();
+  }
+
 }

api/src/main/java/io/kafbat/ui/serdes/ClusterSerdes.java

@@ -64,12 +64,12 @@ public Stream<SerdeInstance> all() {
 
   public SerdeInstance suggestSerdeForSerialize(String topic, Serde.Target type) {
     return findSerdeByPatternsOrDefault(topic, type, s -> s.canSerialize(topic, type))
-        .orElse(serdes.get(StringSerde.name()));
+        .orElse(serdes.get(StringSerde.NAME));
   }
 
   public SerdeInstance suggestSerdeForDeserialize(String topic, Serde.Target type) {
     return findSerdeByPatternsOrDefault(topic, type, s -> s.canDeserialize(topic, type))
-        .orElse(serdes.get(StringSerde.name()));
+        .orElse(serdes.get(StringSerde.NAME));
   }
 
   @Override