BE: Fixes #498 Added sort by number of messages (#1330)

Author: germanosin

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

@@ -7,7 +7,6 @@
 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;
@@ -362,6 +361,10 @@ private Comparator<InternalTopic> getComparatorForTopic(
       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;
     };
   }

api/src/main/java/io/kafbat/ui/mapper/ClusterMapper.java

@@ -51,6 +51,7 @@
 import org.apache.kafka.common.resource.ResourceType;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
+import org.openapitools.jackson.nullable.JsonNullable;
 
 @Mapper(componentModel = "spring")
 public interface ClusterMapper {
@@ -104,6 +105,14 @@ default ConfigSynonymDTO toConfigSynonym(ConfigEntry.ConfigSynonym config) {
 
   TopicDTO toTopic(InternalTopic topic);
 
+  default <T> JsonNullable<T> toJsonNullable(T value) {
+    if (value == null) {
+      return JsonNullable.undefined();
+    } else {
+      return JsonNullable.of(value);
+    }
+  }
+
   PartitionDTO toPartition(InternalPartition topic);
 
   BrokerDTO toBrokerDto(InternalBroker broker);

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

@@ -10,7 +10,6 @@
 import lombok.Data;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.TopicDescription;
-import org.apache.kafka.common.TopicPartition;
 
 @Data
 @Builder(toBuilder = true)
@@ -143,4 +142,16 @@ public static InternalTopic from(TopicDescription topicDescription,
     return topic.build();
   }
 
+  public @Nullable Long getMessagesCount() {
+    Long result = null;
+    if (cleanUpPolicy.equals(CleanupPolicy.DELETE)) {
+      result = 0L;
+      if (partitions != null && !partitions.isEmpty()) {
+        for (InternalPartition partition : partitions.values()) {
+          result += (partition.getOffsetMax() - partition.getOffsetMin());
+        }
+      }
+    }
+    return result;
+  }
 }

contract-typespec/api/topics.tsp

@@ -156,6 +156,7 @@ enum TopicColumnsToSort {
   TOTAL_PARTITIONS,
   REPLICATION_FACTOR,
   SIZE,
+  MESSAGES_COUNT
 }
 
 model Topic {
@@ -170,6 +171,7 @@ model Topic {
   bytesInPerSec?: float64;
   bytesOutPerSec?: float64;
   underReplicatedPartitions?: int32;
+  messagesCount?: int64 | null;
   cleanUpPolicy?: CleanUpPolicy;
   partitions?: Partition[];
 }

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

@@ -2692,6 +2692,7 @@ components:
         - TOTAL_PARTITIONS
         - REPLICATION_FACTOR
         - SIZE
+        - MESSAGES_COUNT
 
     SchemaColumnsToSort:
       type: string
@@ -2747,6 +2748,10 @@ components:
           type: array
           items:
             $ref: "#/components/schemas/Partition"
+        messagesCount:
+          type: integer
+          format: int64
+          nullable: true
       required:
         - name