Refactoring

Author: germanosin

api/build.gradle

@@ -72,9 +72,6 @@ dependencies {
     // CVE Fixes
     implementation libs.apache.commons.compress
     implementation libs.okhttp3.logging.intercepter
-    implementation libs.reactor.netty.http
-    implementation libs.netty.codec.http2
-    // CVE Fixes End
 
     implementation libs.modelcontextprotocol.spring.webflux
     implementation libs.victools.jsonschema.generator

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

@@ -41,7 +41,7 @@ public class ClustersProperties {
   MetricsStorage defaultMetricsStorage = new MetricsStorage();
 
   CacheProperties cache = new CacheProperties();
-  FtsProperties fts = new FtsProperties();
+  ClusterFtsProperties fts = new ClusterFtsProperties();
 
   @Data
   public static class Cluster {
@@ -222,12 +222,21 @@ public static class CacheProperties {
   @NoArgsConstructor
   @AllArgsConstructor
   public static class FtsProperties {
+    boolean ngram = true;
+    int ngramMin = 1;
+    int ngramMax = 4;
+  }
+
+  @Data
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class ClusterFtsProperties {
     boolean enabled = false;
-    boolean topicsNgramEnabled = false;
-    int topicsMinNGram = 3;
-    int topicsMaxNGram = 5;
-    int filterMinNGram = 1;
-    int filterMaxNGram = 4;
+    FtsProperties topics = new FtsProperties(true, 3, 5);
+    FtsProperties schemas = new FtsProperties(true, 1, 4);
+    FtsProperties consumers = new FtsProperties(true, 1, 4);
+    FtsProperties connect = new FtsProperties(true, 1, 4);
+    FtsProperties acl = new FtsProperties(true, 1, 4);
   }
 
   @PostConstruct

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

@@ -214,7 +214,7 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
         .operationName("getSchemas")
         .build();
 
-    ClustersProperties.FtsProperties fts = clustersProperties.getFts();
+    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
 
     return schemaRegistryService
         .getAllSubjectNames(getCluster(clusterName))
@@ -225,9 +225,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.getFilterMinNGram(), fts.getFilterMaxNGram(), fts.isEnabled());
-
+          SchemasFilter filter = new SchemasFilter(subjects, fts.isEnabled(), fts.getSchemas());
           List<String> filteredSubjects = filter.find(search);
 
           var totalPages = (filteredSubjects.size() / pageSize)

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

@@ -38,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;
@@ -188,19 +187,12 @@ public Mono<ResponseEntity<TopicsResponseDTO>> getTopics(String clusterName,
         .flatMap(topics -> {
           int pageSize = perPage != null && perPage > 0 ? perPage : DEFAULT_PAGE_SIZE;
           var topicsToSkip = ((page != null && page > 0 ? page : 1) - 1) * pageSize;
-          ClustersProperties.FtsProperties fts = clustersProperties.getFts();
+          ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
           Comparator<InternalTopic> comparatorForTopic = getComparatorForTopic(orderBy, fts.isEnabled());
           var comparator = sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
               ? comparatorForTopic : comparatorForTopic.reversed();
 
-          List<InternalTopic> filtered = fts.isEnabled() ? topics : topics.stream()
-              .filter(topic -> !topic.isInternal()
-                  || showInternal != null && showInternal)
-              .filter(
-                  topic -> search == null || CI.contains(topic.getName(), search)
-              )
-              .sorted(comparator)
-              .toList();
+          List<InternalTopic> filtered = topics.stream().sorted(comparator).toList();
 
           var totalPages = (filtered.size() / pageSize)
               + (filtered.size() % pageSize == 0 ? 0 : 1);

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,

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

@@ -1,7 +1,5 @@
 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;
@@ -27,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;
@@ -131,16 +128,9 @@ public Mono<ConsumerGroupsPage> getConsumerGroupsPage(
   }
 
   private Collection<ConsumerGroupListing> filterGroups(Collection<ConsumerGroupListing> groups, String search) {
-    if (search == null || search.isBlank()) {
-      return groups;
-    }
-    ClustersProperties.FtsProperties fts = clustersProperties.getFts();
-    if (fts.isEnabled()) {
-      ConsumerGroupFilter filter = new ConsumerGroupFilter(groups, fts.getFilterMinNGram(), fts.getFilterMaxNGram());
-      return filter.find(search);
-    } else {
-      return groups.stream().filter(g -> CI.contains(g.groupId(), search)).toList();
-    }
+    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,

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

@@ -158,21 +158,10 @@ public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
   }
 
   private List<FullConnectorInfoDTO> filterConnectors(List<FullConnectorInfoDTO> connectors, String search) {
-    if (search == null) {
-      return connectors;
-    }
-    ClustersProperties.FtsProperties fts = clustersProperties.getFts();
-    if (fts.isEnabled()) {
-      KafkaConnectNgramFilter filter =
-          new KafkaConnectNgramFilter(connectors, fts.getFilterMinNGram(), fts.getFilterMaxNGram());
-      return filter.find(search);
-    } else {
-      return connectors.stream()
-          .filter(connector -> getStringsForSearch(connector)
-              .anyMatch(string -> CI.contains(string, search)))
-          .toList();
-    }
-
+    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/TopicsService.java

@@ -2,7 +2,6 @@
 
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
-import static org.apache.commons.lang3.Strings.CI;
 
 import com.google.common.collect.Sets;
 import io.kafbat.ui.config.ClustersProperties;
@@ -471,46 +470,29 @@ public Mono<InternalTopic> cloneTopic(
   public Mono<List<InternalTopic>> getTopicsForPagination(KafkaCluster cluster, String search, Boolean showInternal) {
     Statistics stats = statisticsCache.get(cluster);
     ScrapedClusterState clusterState = stats.getClusterState();
-    ClustersProperties.FtsProperties fts = clustersProperties.getFts();
-    Mono<List<String>> topics;
-
-    Map<String, TopicState> topicStates = clusterState.getTopicStates();
-    if (fts.isEnabled() && clusterState.getTopicIndex() != null && search != null && !search.isBlank()) {
-      try {
-        topics = Mono.just(clusterState.getTopicIndex().find(search, showInternal, null));
-      } catch (IOException e) {
-        topics = Mono.error(e);
-      }
-    } else {
-      topics = Mono.just(new ArrayList<>(topicStates.keySet()));
-    }
 
-    return topics.flatMap(lst -> filterExisting(cluster, lst))
-        .flatMapMany(Flux::fromIterable)
-        .map(topicName ->  InternalTopic.from(
-            topicStates.get(topicName).description(),
-            topicStates.get(topicName).configs(),
-            InternalPartitionsOffsets.empty(),
-            stats.getMetrics(),
-            Optional.ofNullable(topicStates.get(topicName))
-                .map(TopicState::segmentStats).orElse(null),
-            Optional.ofNullable(topicStates.get(topicName))
-                .map(TopicState::partitionsSegmentStats).orElse(null),
-            clustersProperties.getInternalTopicPrefix()
-        )).collectList();
+    try {
+      return Mono.just(
+          clusterState.getTopicIndex().find(search, showInternal, null)
+      ).flatMap(lst -> filterExisting(cluster, lst)).map(lst ->
+        lst.stream().map(t -> t.withMetrics(stats.getMetrics())).toList()
+      );
+    } catch (Exception e) {
+      return Mono.error(e);
+    }
   }
 
   public Mono<Map<TopicPartition, List<ProducerState>>> getActiveProducersState(KafkaCluster cluster, String topic) {
     return adminClientService.get(cluster)
         .flatMap(ac -> ac.getActiveProducersState(topic));
   }
 
-  private Mono<List<String>> filterExisting(KafkaCluster cluster, Collection<String> topics) {
+  private Mono<List<InternalTopic>> filterExisting(KafkaCluster cluster, Collection<InternalTopic> topics) {
     return adminClientService.get(cluster)
         .flatMap(ac -> ac.listTopics(true))
-        .map(existing -> existing
+        .map(existing -> topics
             .stream()
-            .filter(topics::contains)
+            .filter(s -> existing.contains(s.getName()))
             .collect(toList()));
   }
 

api/src/main/java/io/kafbat/ui/service/acl/AclsService.java

@@ -34,7 +34,6 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.acl.AccessControlEntry;
 import org.apache.kafka.common.acl.AclBinding;
-import org.apache.kafka.common.acl.AclBindingFilter;
 import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.resource.Resource;
 import org.apache.kafka.common.resource.ResourcePattern;
@@ -84,18 +83,9 @@ public Flux<AclBinding> listAcls(KafkaCluster cluster, ResourcePatternFilter fil
   }
 
   private List<AclBinding> filter(List<AclBinding> acls, String principalSearch) {
-    if (principalSearch == null) {
-      return acls;
-    }
-    ClustersProperties.FtsProperties fts = clustersProperties.getFts();
-    if (fts.isEnabled()) {
-      AclBindingNgramFilter filter = new AclBindingNgramFilter(acls, fts.getFilterMinNGram(), fts.getFilterMaxNGram());
-      return filter.find(principalSearch);
-    } else {
-      return acls.stream()
-          .filter(acl -> acl.entry().principal().contains(principalSearch))
-          .toList();
-    }
+    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
+    AclBindingNgramFilter filter = new AclBindingNgramFilter(acls, fts.isEnabled(), fts.getAcl());
+    return filter.find(principalSearch);
   }
 
   public Mono<String> getAclAsCsvString(KafkaCluster cluster) {

api/src/main/java/io/kafbat/ui/service/index/AclBindingNgramFilter.java

@@ -1,5 +1,6 @@
 package io.kafbat.ui.service.index;
 
+import io.kafbat.ui.config.ClustersProperties;
 import java.util.Collection;
 import java.util.List;
 import org.apache.kafka.common.acl.AclBinding;
@@ -10,11 +11,14 @@ public class AclBindingNgramFilter extends NgramFilter<AclBinding> {
   private final List<Tuple2<List<String>, AclBinding>> bindings;
 
   public AclBindingNgramFilter(Collection<AclBinding> bindings) {
-    this(bindings, 1, 4);
+    this(bindings, true, new ClustersProperties.FtsProperties(true, 1, 4));
   }
 
-  public AclBindingNgramFilter(Collection<AclBinding> bindings, int minNGram, int maxNGram) {
-    super(minNGram, maxNGram);
+  public AclBindingNgramFilter(
+      Collection<AclBinding> bindings,
+      boolean enabled,
+      ClustersProperties.FtsProperties properties) {
+    super(properties, enabled);
     this.bindings = bindings.stream().map(g -> Tuples.of(List.of(g.entry().principal()), g)).toList();
   }