Merge branch 'main' into issues/1312

Author: germanosin

.github/dependabot.yml

@@ -20,10 +20,19 @@ updates:
       update-types:
         - "patch"
         - "minor"
+      exclude-patterns:
+          - "org.springframework.boot:*"
+          - "io.spring.dependency-management"
+          # All netty references are temporary overwrites that must be set carefully
+          # We do not need dependabot to send pull requests
+          - "io.netty:*"    
     other-dependencies:
       exclude-patterns:
         - "org.springframework.boot:*"
         - "io.spring.dependency-management"
+        # All netty references are temporary overwrites that must be set carefully
+        # We do not need dependabot to send pull requests
+        - "io.netty:*"
       patterns:
         - "*"
       update-types:

.github/workflows/e2e-playwright-manual.yml

@@ -0,0 +1,22 @@
+name: "E2E: Playwright Manual run"
+on:
+  workflow_dispatch:
+    inputs:
+      sha:
+        description: Commit to run on
+        required: true
+        type: string
+
+permissions:
+  contents: read
+  checks: write
+  statuses: write
+
+jobs:
+  build-and-test:
+    uses: ./.github/workflows/e2e-playwright-run.yml
+    secrets: inherit
+    with:
+      sha: ${{ inputs.sha }}
+
+

.github/workflows/e2e-playwright-pr.yml

@@ -0,0 +1,27 @@
+name: "Playwright E2E: PR tests"
+on:
+  pull_request:
+    types: [ "opened", "reopened", "synchronize" ]
+    paths:
+      - "build.gradle"
+      - "gradle.properties"
+      - "settings.gradle"
+      - "gradle/libs.versions.toml"
+
+      - "contract/**"
+      - "api/**"
+      - "serde-api/**"
+      - "frontend/**"
+      - "e2e-playwright/**"
+
+permissions:
+  contents: read
+  checks: write
+  statuses: write
+
+jobs:
+  build-and-test:
+    uses: ./.github/workflows/e2e-playwright-run.yml
+    secrets: inherit
+    with:
+      sha: ${{ github.event.pull_request.head.sha }}

.github/workflows/e2e-playwright.yml renamed to .github/workflows/e2e-playwright-run.yml

@@ -1,9 +1,11 @@
-name: "E2E: Playwright Manual run"
+name: "E2E: Playwright"
+
 on:
-  workflow_dispatch:
-    sha:
-      required: true
-      type: string
+  workflow_call:
+    inputs:
+      sha:
+        required: true
+        type: string
 
 permissions:
   contents: read
@@ -30,6 +32,8 @@ jobs:
         uses: actions/setup-node@v3
         with:
           node-version: 18
+          cache-dependency-path: ./e2e-playwright/package-lock.json
+          cache: 'npm'
 
       - name: Install NPM dependencies
         working-directory: ./e2e-playwright

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'
@@ -68,8 +72,13 @@ dependencies {
     // CVE Fixes
     implementation libs.apache.commons.compress
     implementation libs.okhttp3.logging.intercepter
+    // START Fixes https://www.cve.org/CVERecord?id=CVE-2025-58056 and https://www.cve.org/CVERecord?id=CVE-2025-58057
+    implementation libs.netty.codec
+    implementation libs.netty.codec.http
+    // 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,27 +1,32 @@
 package io.kafbat.ui.controller;
 
-import static org.apache.commons.lang3.Strings.CI;
-
 import io.kafbat.ui.api.SchemasApi;
+import io.kafbat.ui.api.model.SchemaColumnsToSort;
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.mapper.KafkaSrMapper;
 import io.kafbat.ui.mapper.KafkaSrMapperImpl;
 import io.kafbat.ui.model.CompatibilityCheckResponseDTO;
 import io.kafbat.ui.model.CompatibilityLevelDTO;
+import io.kafbat.ui.model.InternalTopic;
 import io.kafbat.ui.model.KafkaCluster;
 import io.kafbat.ui.model.NewSchemaSubjectDTO;
+import io.kafbat.ui.model.SchemaColumnsToSortDTO;
 import io.kafbat.ui.model.SchemaSubjectDTO;
 import io.kafbat.ui.model.SchemaSubjectsResponseDTO;
+import io.kafbat.ui.model.SortOrderDTO;
 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.SchemaRegistryService.SubjectWithCompatibilityLevel;
+import io.kafbat.ui.service.index.SchemasFilter;
 import io.kafbat.ui.service.mcp.McpTool;
+import java.util.Comparator;
 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 +43,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) {
@@ -208,12 +214,16 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
                                                                     @Valid Integer pageNum,
                                                                     @Valid Integer perPage,
                                                                     @Valid String search,
+                                                                    SchemaColumnsToSortDTO orderBy,
+                                                                    SortOrderDTO sortOrder,
                                                                     ServerWebExchange serverWebExchange) {
     var context = AccessContext.builder()
         .cluster(clusterName)
         .operationName("getSchemas")
         .build();
 
+    ClustersProperties.ClusterFtsProperties fts = clustersProperties.getFts();
+
     return schemaRegistryService
         .getAllSubjectNames(getCluster(clusterName))
         .flatMapIterable(l -> l)
@@ -222,23 +232,72 @@ 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()
-              .skip(subjectToSkip)
-              .limit(pageSize)
-              .toList();
-          return schemaRegistryService.getAllLatestVersionSchemas(getCluster(clusterName), subjectsToRender)
-              .map(subjs -> subjs.stream().map(kafkaSrMapper::toDto).toList())
+
+          List<String> subjectsToRetrieve;
+          boolean paginate = true;
+          var schemaComparator = getComparatorForSchema(orderBy);
+          final Comparator<SubjectWithCompatibilityLevel> comparator =
+              sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
+                  ? schemaComparator : schemaComparator.reversed();
+          if (orderBy == null || SchemaColumnsToSortDTO.SUBJECT.equals(orderBy)) {
+            if (SortOrderDTO.DESC.equals(sortOrder)) {
+              filteredSubjects.sort(Comparator.reverseOrder());
+            }
+            subjectsToRetrieve = filteredSubjects.stream()
+                .skip(subjectToSkip)
+                .limit(pageSize)
+                .toList();
+            paginate = false;
+          } else {
+            subjectsToRetrieve = filteredSubjects;
+          }
+
+          final boolean shouldPaginate = paginate;
+
+          return schemaRegistryService.getAllLatestVersionSchemas(getCluster(clusterName), subjectsToRetrieve, pageSize)
+              .map(subjs ->
+                  paginateSchemas(subjs, comparator, shouldPaginate, pageSize, subjectToSkip)
+              ).map(subjs -> subjs.stream().map(kafkaSrMapper::toDto).toList())
               .map(subjs -> new SchemaSubjectsResponseDTO().pageCount(totalPages).schemas(subjs));
         }).map(ResponseEntity::ok)
         .doOnEach(sig -> audit(context, sig));
   }
 
+  private List<SubjectWithCompatibilityLevel> paginateSchemas(
+      List<SubjectWithCompatibilityLevel> subjects,
+      Comparator<SubjectWithCompatibilityLevel> comparator,
+      boolean paginate,
+      int pageSize,
+      int subjectToSkip) {
+    subjects.sort(comparator);
+    if (paginate) {
+      return subjects.subList(subjectToSkip, Math.min(subjectToSkip + pageSize, subjects.size()));
+    } else {
+      return subjects;
+    }
+  }
+
+  private Comparator<SubjectWithCompatibilityLevel> getComparatorForSchema(
+      @Valid SchemaColumnsToSortDTO orderBy) {
+    var defaultComparator = Comparator.comparing(SubjectWithCompatibilityLevel::getSubject);
+    if (orderBy == null) {
+      return defaultComparator;
+    }
+    return switch (orderBy) {
+      case SUBJECT -> Comparator.comparing(SubjectWithCompatibilityLevel::getSubject);
+      case ID -> Comparator.comparing(SubjectWithCompatibilityLevel::getId);
+      case TYPE -> Comparator.comparing(SubjectWithCompatibilityLevel::getSchemaType);
+      case COMPATIBILITY -> Comparator.comparing(SubjectWithCompatibilityLevel::getCompatibility);
+      default -> defaultComparator;
+    };
+  }
+
   @Override
   public Mono<ResponseEntity<Void>> updateGlobalSchemaCompatibilityLevel(
       String clusterName, @Valid Mono<CompatibilityLevelDTO> compatibilityLevelMono,

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

@@ -7,9 +7,9 @@
 import static io.kafbat.ui.model.rbac.permission.TopicAction.EDIT;
 import static io.kafbat.ui.model.rbac.permission.TopicAction.VIEW;
 import static java.util.stream.Collectors.toList;
-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 +37,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 +54,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 +181,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,16 +348,23 @@ 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) {
       case TOTAL_PARTITIONS -> Comparator.comparing(InternalTopic::getPartitionCount);
       case OUT_OF_SYNC_REPLICAS -> Comparator.comparing(t -> t.getReplicas() - t.getInSyncReplicas());
       case REPLICATION_FACTOR -> Comparator.comparing(InternalTopic::getReplicationFactor);
       case SIZE -> Comparator.comparing(InternalTopic::getSegmentSize);
+      case MESSAGES_COUNT ->  Comparator.comparing(
+          InternalTopic::getMessagesCount,
+          Comparator.nullsLast(Long::compareTo)
+      );
       default -> defaultComparator;
     };
   }