BE: Added schemas ordering backend

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

@@ -1,21 +1,27 @@
 package io.kafbat.ui.controller;
 
 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;
@@ -208,6 +214,8 @@ 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)
@@ -230,17 +238,66 @@ public Mono<ResponseEntity<SchemaSubjectsResponseDTO>> getSchemas(String cluster
 
           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/service/SchemaRegistryService.java

@@ -47,9 +47,10 @@ private ReactiveFailover<KafkaSrClientApi> api(KafkaCluster cluster) {
   }
 
   public Mono<List<SubjectWithCompatibilityLevel>> getAllLatestVersionSchemas(KafkaCluster cluster,
-                                                                              List<String> subjects) {
+                                                                              List<String> subjects,
+                                                                              int pageSize) {
     return Flux.fromIterable(subjects)
-        .concatMap(subject -> getLatestSchemaVersionBySubject(cluster, subject))
+        .flatMap(subject -> getLatestSchemaVersionBySubject(cluster, subject), pageSize)
         .collect(Collectors.toList());
   }
 

api/src/test/java/io/kafbat/ui/service/SchemaRegistryPaginationTest.java

@@ -1,6 +1,7 @@
 package io.kafbat.ui.service;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.mock;
@@ -9,15 +10,22 @@
 import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.controller.SchemasController;
 import io.kafbat.ui.model.KafkaCluster;
+import io.kafbat.ui.model.SchemaColumnsToSortDTO;
 import io.kafbat.ui.model.SchemaSubjectDTO;
+import io.kafbat.ui.model.SortOrderDTO;
+import io.kafbat.ui.service.SchemaRegistryService.SubjectWithCompatibilityLevel;
 import io.kafbat.ui.service.audit.AuditService;
 import io.kafbat.ui.sr.model.Compatibility;
 import io.kafbat.ui.sr.model.SchemaSubject;
+import io.kafbat.ui.sr.model.SchemaType;
 import io.kafbat.ui.util.AccessControlServiceMock;
 import io.kafbat.ui.util.ReactiveFailover;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -30,19 +38,31 @@ class SchemaRegistryPaginationTest {
   private SchemasController controller;
 
   private void init(List<String> subjects) {
+    initWithData(subjects.stream().map(s ->
+        new SubjectWithCompatibilityLevel(
+            new SchemaSubject().subject(s),
+            Compatibility.FULL
+        )
+    ).toList());
+  }
+
+  private void initWithData(List<SubjectWithCompatibilityLevel> subjects) {
     ClustersStorage clustersStorage = Mockito.mock(ClustersStorage.class);
     when(clustersStorage.getClusterByName(isA(String.class)))
         .thenReturn(Optional.of(buildKafkaCluster(LOCAL_KAFKA_CLUSTER_NAME)));
 
+    Map<String, SubjectWithCompatibilityLevel> subjectsMap = subjects.stream().collect(Collectors.toMap(
+        SubjectWithCompatibilityLevel::getSubject,
+        Function.identity()
+    ));
+
     SchemaRegistryService schemaRegistryService = Mockito.mock(SchemaRegistryService.class);
     when(schemaRegistryService.getAllSubjectNames(isA(KafkaCluster.class)))
-                .thenReturn(Mono.just(subjects));
+        .thenReturn(Mono.just(subjects.stream().map(SubjectWithCompatibilityLevel::getSubject).toList()));
     when(schemaRegistryService
-            .getAllLatestVersionSchemas(isA(KafkaCluster.class), anyList())).thenCallRealMethod();
+        .getAllLatestVersionSchemas(isA(KafkaCluster.class), anyList(), anyInt())).thenCallRealMethod();
     when(schemaRegistryService.getLatestSchemaVersionBySubject(isA(KafkaCluster.class), isA(String.class)))
-            .thenAnswer(a -> Mono.just(
-                new SchemaRegistryService.SubjectWithCompatibilityLevel(
-                    new SchemaSubject().subject(a.getArgument(1)), Compatibility.FULL)));
+        .thenAnswer(a -> Mono.just(subjectsMap.get(a.getArgument(1))));
 
     this.controller = new SchemasController(schemaRegistryService, new ClustersProperties());
     this.controller.setAccessControlService(new AccessControlServiceMock().getMock());
@@ -59,7 +79,7 @@ void shouldListFirst25andThen10Schemas() {
                     .toList()
     );
     var schemasFirst25 = controller.getSchemas(LOCAL_KAFKA_CLUSTER_NAME,
-            null, null, null, null).block();
+            null, null, null, null, null, null).block();
     assertThat(schemasFirst25).isNotNull();
     assertThat(schemasFirst25.getBody()).isNotNull();
     assertThat(schemasFirst25.getBody().getPageCount()).isEqualTo(4);
@@ -68,7 +88,7 @@ void shouldListFirst25andThen10Schemas() {
             .isSortedAccordingTo(Comparator.comparing(SchemaSubjectDTO::getSubject));
 
     var schemasFirst10 = controller.getSchemas(LOCAL_KAFKA_CLUSTER_NAME,
-            null, 10, null, null).block();
+            null, 10, null, null, null, null).block();
 
     assertThat(schemasFirst10).isNotNull();
     assertThat(schemasFirst10.getBody()).isNotNull();
@@ -87,7 +107,7 @@ void shouldListSchemasContaining_1() {
                       .toList()
     );
     var schemasSearch7 = controller.getSchemas(LOCAL_KAFKA_CLUSTER_NAME,
-            null, null, "1", null).block();
+            null, null, "1", null, null, null).block();
     assertThat(schemasSearch7).isNotNull();
     assertThat(schemasSearch7.getBody()).isNotNull();
     assertThat(schemasSearch7.getBody().getPageCount()).isEqualTo(1);
@@ -103,7 +123,7 @@ void shouldCorrectlyHandleNonPositivePageNumberAndPageSize() {
                         .toList()
     );
     var schemas = controller.getSchemas(LOCAL_KAFKA_CLUSTER_NAME,
-            0, -1, null, null).block();
+            0, -1, null, null, null, null).block();
 
     assertThat(schemas).isNotNull();
     assertThat(schemas.getBody()).isNotNull();
@@ -122,7 +142,7 @@ void shouldCalculateCorrectPageCountForNonDivisiblePageSize() {
     );
 
     var schemas = controller.getSchemas(LOCAL_KAFKA_CLUSTER_NAME,
-            4, 33, null, null).block();
+            4, 33, null, null, null, null).block();
 
     assertThat(schemas).isNotNull();
     assertThat(schemas.getBody()).isNotNull();
@@ -138,4 +158,39 @@ private KafkaCluster buildKafkaCluster(String clusterName) {
             .schemaRegistryClient(mock(ReactiveFailover.class))
             .build();
   }
+
+  @Test
+  void shouldOrderByAndPaginate() {
+    List<SubjectWithCompatibilityLevel> schemas = IntStream.rangeClosed(1, 100)
+        .boxed()
+        .map(num -> new
+                SubjectWithCompatibilityLevel(
+                new SchemaSubject()
+                    .subject("subject" + num)
+                    .schemaType(SchemaType.AVRO)
+                    .id(num),
+                Compatibility.FULL
+            )
+        ).toList();
+
+    initWithData(schemas);
+
+    var schemasFirst25 = controller.getSchemas(LOCAL_KAFKA_CLUSTER_NAME,
+        null, null, null,
+        SchemaColumnsToSortDTO.ID, SortOrderDTO.DESC, null
+    ).block();
+
+    List<String> last25OrderedById = schemas.stream()
+        .sorted(Comparator.comparing(SubjectWithCompatibilityLevel::getId).reversed())
+        .map(SubjectWithCompatibilityLevel::getSubject)
+        .limit(25)
+        .toList();
+
+    assertThat(schemasFirst25).isNotNull();
+    assertThat(schemasFirst25.getBody()).isNotNull();
+    assertThat(schemasFirst25.getBody().getPageCount()).isEqualTo(4);
+    assertThat(schemasFirst25.getBody().getSchemas()).hasSize(25);
+    assertThat(schemasFirst25.getBody().getSchemas().stream().map(SchemaSubjectDTO::getSubject).toList())
+        .isEqualTo(last25OrderedById);
+  }
 }

contract-typespec/api/schemas.tsp

@@ -1,5 +1,6 @@
 import "@typespec/openapi";
 import "./responses.tsp";
+import "./models.tsp";
 
 namespace Api;
 
@@ -26,6 +27,8 @@ interface SchemasApi {
     @query page?: int32,
     @query perPage?: int32,
     @query search?: string,
+    @query orderBy?: SchemaColumnsToSort,
+    @query sortOrder?: SortOrder,
   ): SchemaSubjectsResponse;
 
   @delete
@@ -172,3 +175,10 @@ model SchemaSubjectsResponse {
   pageCount?: int32;
   schemas?: SchemaSubject[];
 }
+
+enum SchemaColumnsToSort {
+  SUBJECT,
+  ID,
+  TYPE,
+  COMPATIBILITY,
+}

contract-typespec/api/topics.tsp

@@ -1,5 +1,6 @@
 import "@typespec/openapi";
 import "./responses.tsp";
+import "./models.tsp";
 
 namespace Api;
 

contract/src/main/resources/swagger/kafbat-ui-api.yaml

@@ -1208,6 +1208,16 @@ paths:
           required: false
           schema:
             type: string
+        - name: orderBy
+          in: query
+          required: false
+          schema:
+            $ref: '#/components/schemas/SchemaColumnsToSort'
+        - name: sortOrder
+          in: query
+          required: false
+          schema:
+            $ref: '#/components/schemas/SortOrder'
       responses:
         200:
           description: OK
@@ -2683,6 +2693,14 @@ components:
         - REPLICATION_FACTOR
         - SIZE
 
+    SchemaColumnsToSort:
+      type: string
+      enum:
+        - SUBJECT
+        - ID
+        - TYPE
+        - COMPATIBILITY
+
     ConnectorColumnsToSort:
       type: string
       enum: