BE: Full text search support (#1267)

Author: germanosin

api/build.gradle

@@ -54,6 +54,10 @@ dependencies {
     antlr libs.antlr
     implementation libs.antlr.runtime
 
+    implementation libs.lucene
+    implementation libs.lucene.queryparser
+    implementation libs.lucene.analysis.common
+
     implementation libs.opendatadiscovery.oddrn
     implementation(libs.opendatadiscovery.client) {
         exclude group: 'org.springframework.boot', module: 'spring-boot-starter-webflux'
@@ -74,6 +78,7 @@ dependencies {
     // END Fixes https://www.cve.org/CVERecord?id=CVE-2025-58056 and https://www.cve.org/CVERecord?id=CVE-2025-58057
     // CVE Fixes End
 
+
     implementation libs.modelcontextprotocol.spring.webflux
     implementation libs.victools.jsonschema.generator
 

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

@@ -41,6 +41,7 @@ public class ClustersProperties {
   MetricsStorage defaultMetricsStorage = new MetricsStorage();
 
   CacheProperties cache = new CacheProperties();
+  ClusterFtsProperties fts = new ClusterFtsProperties();
 
   @Data
   public static class Cluster {
@@ -217,6 +218,25 @@ public static class CacheProperties {
     Duration connectClusterCacheExpiry = Duration.ofHours(24);
   }
 
+  @Data
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class NgramProperties {
+    int ngramMin = 1;
+    int ngramMax = 4;
+  }
+
+  @Data
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class ClusterFtsProperties {
+    boolean enabled = 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);
+  }
+
   @PostConstruct
   public void validateAndSetDefaults() {
     if (clusters != null) {

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

@@ -1,8 +1,7 @@
 package io.kafbat.ui.controller;
 
-import static org.apache.commons.lang3.Strings.CI;
-
 import io.kafbat.ui.api.SchemasApi;
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.mapper.KafkaSrMapper;
 import io.kafbat.ui.mapper.KafkaSrMapperImpl;
@@ -15,13 +14,13 @@
 import io.kafbat.ui.model.rbac.AccessContext;
 import io.kafbat.ui.model.rbac.permission.SchemaAction;
 import io.kafbat.ui.service.SchemaRegistryService;
+import io.kafbat.ui.service.index.SchemasFilter;
 import io.kafbat.ui.service.mcp.McpTool;
 import java.util.List;
 import java.util.Map;
 import javax.validation.Valid;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.server.ServerWebExchange;
@@ -38,6 +37,7 @@ public class SchemasController extends AbstractController implements SchemasApi,
   private final KafkaSrMapper kafkaSrMapper = new KafkaSrMapperImpl();
 
   private final SchemaRegistryService schemaRegistryService;
+  private final ClustersProperties clustersProperties;
 
   @Override
   protected KafkaCluster getCluster(String clusterName) {
@@ -214,6 +214,8 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
         .operationName("getSchemas")
         .build();
 
+    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
+
     return schemaRegistryService
         .getAllSubjectNames(getCluster(clusterName))
         .flatMapIterable(l -> l)
@@ -222,10 +224,10 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
         .flatMap(subjects -> {
           int pageSize = perPage != null && perPage > 0 ? perPage : DEFAULT_PAGE_SIZE;
           int subjectToSkip = ((pageNum != null && pageNum > 0 ? pageNum : 1) - 1) * pageSize;
-          List<String> filteredSubjects = subjects
-              .stream()
-              .filter(subj -> search == null || CI.contains(subj, search))
-              .sorted().toList();
+
+          SchemasFilter filter = new SchemasFilter(subjects, fts.isEnabled(), fts.getSchemas());
+          List<String> filteredSubjects = filter.find(search);
+
           var totalPages = (filteredSubjects.size() / pageSize)
               + (filteredSubjects.size() % pageSize == 0 ? 0 : 1);
           List<String> subjectsToRender = filteredSubjects.stream()

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

@@ -10,6 +10,7 @@
 import static org.apache.commons.lang3.Strings.CI;
 
 import io.kafbat.ui.api.TopicsApi;
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.mapper.ClusterMapper;
 import io.kafbat.ui.model.InternalTopic;
 import io.kafbat.ui.model.InternalTopicConfig;
@@ -37,7 +38,6 @@
 import javax.validation.Valid;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.RestController;
@@ -55,6 +55,7 @@ public class TopicsController extends AbstractController implements TopicsApi, M
   private final TopicsService topicsService;
   private final TopicAnalysisService topicAnalysisService;
   private final ClusterMapper clusterMapper;
+  private final ClustersProperties clustersProperties;
 
   @Override
   public Mono<ResponseEntity<TopicDTO>> createTopic(
@@ -181,23 +182,23 @@ public Mono<ResponseEntity<TopicsResponseDTO>> getTopics(String clusterName,
         .operationName("getTopics")
         .build();
 
-    return topicsService.getTopicsForPagination(getCluster(clusterName))
+    return topicsService.getTopicsForPagination(getCluster(clusterName), search, showInternal)
         .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());
           var comparator = sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
-              ? getComparatorForTopic(orderBy) : getComparatorForTopic(orderBy).reversed();
-          List<InternalTopic> filtered = topics.stream()
-              .filter(topic -> !topic.isInternal()
-                  || showInternal != null && showInternal)
-              .filter(topic -> search == null || CI.contains(topic.getName(), search))
-              .sorted(comparator)
-              .toList();
+              ? comparatorForTopic : comparatorForTopic.reversed();
+
+          List<InternalTopic> filtered = topics.stream().sorted(comparator).toList();
+
           var totalPages = (filtered.size() / pageSize)
               + (filtered.size() % pageSize == 0 ? 0 : 1);
 
           List<String> topicsPage = filtered.stream()
+              .filter(t -> !t.isInternal() || showInternal != null && showInternal)
               .skip(topicsToSkip)
               .limit(pageSize)
               .map(InternalTopic::getName)
@@ -348,9 +349,12 @@ public Mono<ResponseEntity<Flux<TopicProducerStateDTO>>> getActiveProducerStates
   }
 
   private Comparator<InternalTopic> getComparatorForTopic(
-      TopicColumnsToSortDTO orderBy) {
+      TopicColumnsToSortDTO orderBy,
+      boolean ftsEnabled) {
     var defaultComparator = Comparator.comparing(InternalTopic::getName);
-    if (orderBy == null) {
+    if (orderBy == null && ftsEnabled) {
+      return  (o1, o2) -> 0;
+    } else if (orderBy == null) {
       return defaultComparator;
     }
     return switch (orderBy) {

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

@@ -38,6 +38,16 @@ public class InternalTopic {
   private final long segmentSize;
   private final long segmentCount;
 
+
+  public InternalTopic withMetrics(Metrics metrics) {
+    var builder = toBuilder();
+    if (metrics != null) {
+      builder.bytesInPerSec(metrics.getIoRates().topicBytesInPerSec().get(this.name));
+      builder.bytesOutPerSec(metrics.getIoRates().topicBytesOutPerSec().get(this.name));
+    }
+    return builder.build();
+  }
+
   public static InternalTopic from(TopicDescription topicDescription,
                                    List<ConfigEntry> configs,
                                    InternalPartitionsOffsets partitionsOffsets,
@@ -113,8 +123,10 @@ public static InternalTopic from(TopicDescription topicDescription,
           topic.segmentSize(stats.getSegmentSize());
         });
 
-    topic.bytesInPerSec(metrics.getIoRates().topicBytesInPerSec().get(topicDescription.name()));
-    topic.bytesOutPerSec(metrics.getIoRates().topicBytesOutPerSec().get(topicDescription.name()));
+    if (metrics != null) {
+      topic.bytesInPerSec(metrics.getIoRates().topicBytesInPerSec().get(topicDescription.name()));
+      topic.bytesOutPerSec(metrics.getIoRates().topicBytesOutPerSec().get(topicDescription.name()));
+    }
 
     topic.topicConfigs(
         configs.stream().map(InternalTopicConfig::from).collect(Collectors.toList()));

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

@@ -11,7 +11,7 @@
 
 @Value
 @Builder(toBuilder = true)
-public class Statistics {
+public class Statistics implements AutoCloseable {
   ServerStatusDTO status;
   Throwable lastKafkaException;
   String version;
@@ -46,4 +46,11 @@ public Stream<TopicDescription> topicDescriptions() {
   public Statistics withClusterState(UnaryOperator<ScrapedClusterState> stateUpdate) {
     return toBuilder().clusterState(stateUpdate.apply(clusterState)).build();
   }
+
+  @Override
+  public void close() throws Exception {
+    if (clusterState != null) {
+      clusterState.close();
+    }
+  }
 }

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

@@ -1,15 +1,15 @@
 package io.kafbat.ui.service;
 
-import static org.apache.commons.lang3.Strings.CI;
-
 import com.google.common.collect.Streams;
 import com.google.common.collect.Table;
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.emitter.EnhancedConsumer;
 import io.kafbat.ui.model.ConsumerGroupOrderingDTO;
 import io.kafbat.ui.model.InternalConsumerGroup;
 import io.kafbat.ui.model.InternalTopicConsumerGroup;
 import io.kafbat.ui.model.KafkaCluster;
 import io.kafbat.ui.model.SortOrderDTO;
+import io.kafbat.ui.service.index.ConsumerGroupFilter;
 import io.kafbat.ui.service.rbac.AccessControlService;
 import io.kafbat.ui.util.ApplicationMetrics;
 import io.kafbat.ui.util.KafkaClientSslPropertiesUtil;
@@ -25,7 +25,6 @@
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.admin.ConsumerGroupDescription;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.apache.kafka.clients.admin.OffsetSpec;
@@ -41,6 +40,7 @@ public class ConsumerGroupService {
 
   private final AdminClientService adminClientService;
   private final AccessControlService accessControlService;
+  private final ClustersProperties clustersProperties;
 
   private Mono<List<InternalConsumerGroup>> getConsumerGroups(
       ReactiveAdminClient ac,
@@ -114,11 +114,7 @@ public Mono<ConsumerGroupsPage> getConsumerGroupsPage(
       SortOrderDTO sortOrderDto) {
     return adminClientService.get(cluster).flatMap(ac ->
         ac.listConsumerGroups()
-            .map(listing -> search == null
-                ? listing
-                : listing.stream()
-                .filter(g -> CI.contains(g.groupId(), search))
-                .toList()
+            .map(listing -> filterGroups(listing, search)
             )
             .flatMapIterable(lst -> lst)
             .filterWhen(cg -> accessControlService.isConsumerGroupAccessible(cg.groupId(), cluster.getName()))
@@ -131,6 +127,12 @@ public Mono<ConsumerGroupsPage> getConsumerGroupsPage(
                             (allGroups.size() / perPage) + (allGroups.size() % perPage == 0 ? 0 : 1))))));
   }
 
+  private Collection<ConsumerGroupListing> filterGroups(Collection<ConsumerGroupListing> groups, String search) {
+    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
+    ConsumerGroupFilter filter = new ConsumerGroupFilter(groups, fts.isEnabled(), fts.getConsumers());
+    return filter.find(search, false);
+  }
+
   private Mono<List<ConsumerGroupDescription>> loadSortedDescriptions(ReactiveAdminClient ac,
                                                                       List<ConsumerGroupListing> groups,
                                                                       int pageNum,

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

@@ -27,6 +27,7 @@
 import io.kafbat.ui.model.NewConnectorDTO;
 import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.connect.InternalConnectorInfo;
+import io.kafbat.ui.service.index.KafkaConnectNgramFilter;
 import io.kafbat.ui.util.ReactiveFailover;
 import jakarta.validation.Valid;
 import java.util.List;
@@ -151,15 +152,16 @@ public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
                             .topics(tuple.getT4().getTopics())
                             .build())))
         .map(kafkaConnectMapper::fullConnectorInfo)
-        .filter(matchesSearchTerm(search));
+        .collectList()
+        .map(lst -> filterConnectors(lst, search))
+        .flatMapMany(Flux::fromIterable);
   }
 
-  private Predicate<FullConnectorInfoDTO> matchesSearchTerm(@Nullable final String search) {
-    if (search == null) {
-      return c -> true;
-    }
-    return connector -> getStringsForSearch(connector)
-        .anyMatch(string -> CI.contains(string, search));
+  private List<FullConnectorInfoDTO> filterConnectors(List<FullConnectorInfoDTO> connectors, String search) {
+    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
+    KafkaConnectNgramFilter filter =
+        new KafkaConnectNgramFilter(connectors, fts.isEnabled(), fts.getConnect());
+    return filter.find(search);
   }
 
   private Stream<String> getStringsForSearch(FullConnectorInfoDTO fullConnectorInfo) {

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

@@ -1,16 +1,20 @@
 package io.kafbat.ui.service;
 
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.model.InternalPartitionsOffsets;
 import io.kafbat.ui.model.KafkaCluster;
+import io.kafbat.ui.model.ServerStatusDTO;
 import io.kafbat.ui.model.Statistics;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.springframework.stereotype.Component;
 
+@Slf4j
 @Component
 public class StatisticsCache {
 
@@ -28,12 +32,22 @@ public synchronized void replace(KafkaCluster c, Statistics stats) {
   public synchronized void update(KafkaCluster c,
                                   Map<String, TopicDescription> descriptions,
                                   Map<String, List<ConfigEntry>> configs,
-                                  InternalPartitionsOffsets partitionsOffsets) {
+                                  InternalPartitionsOffsets partitionsOffsets,
+                                  ClustersProperties clustersProperties) {
     var stats = get(c);
     replace(
         c,
-        stats.withClusterState(s -> s.updateTopics(descriptions, configs, partitionsOffsets))
+        stats.withClusterState(s ->
+            s.updateTopics(descriptions, configs, partitionsOffsets, clustersProperties)
+        )
     );
+    try {
+      if (!stats.getStatus().equals(ServerStatusDTO.INITIALIZING)) {
+        stats.close();
+      }
+    } catch (Exception e) {
+      log.error("Error closing cluster {} stats", c.getName(), e);
+    }
   }
 
   public synchronized void onTopicDelete(KafkaCluster c, String topic) {

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

@@ -2,6 +2,7 @@
 
 import static io.kafbat.ui.service.ReactiveAdminClient.ClusterDescription;
 
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.model.ClusterFeature;
 import io.kafbat.ui.model.KafkaCluster;
 import io.kafbat.ui.model.Metrics;
@@ -22,6 +23,7 @@ public class StatisticsService {
   private final AdminClientService adminClientService;
   private final FeatureService featureService;
   private final StatisticsCache cache;
+  private final ClustersProperties clustersProperties;
 
   public Mono<Statistics> updateCache(KafkaCluster c) {
     return getStatistics(c).doOnSuccess(m -> cache.replace(c, m));
@@ -62,7 +64,7 @@ private Statistics createStats(ClusterDescription description,
 
   private Mono<ScrapedClusterState> loadClusterState(ClusterDescription clusterDescription,
                                                      ReactiveAdminClient ac) {
-    return ScrapedClusterState.scrape(clusterDescription, ac);
+    return ScrapedClusterState.scrape(clusterDescription, ac, clustersProperties);
   }
 
   private Mono<Metrics> scrapeMetrics(KafkaCluster cluster,