diff --git a/api/src/main/java/io/kafbat/ui/controller/TopicsController.java b/api/src/main/java/io/kafbat/ui/controller/TopicsController.java index 4ad3f11e9..39885d593 100644 --- a/api/src/main/java/io/kafbat/ui/controller/TopicsController.java +++ b/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 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; }; } diff --git a/api/src/main/java/io/kafbat/ui/mapper/ClusterMapper.java b/api/src/main/java/io/kafbat/ui/mapper/ClusterMapper.java index df3051af3..24ef65991 100644 --- a/api/src/main/java/io/kafbat/ui/mapper/ClusterMapper.java +++ b/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 JsonNullable toJsonNullable(T value) { + if (value == null) { + return JsonNullable.undefined(); + } else { + return JsonNullable.of(value); + } + } + PartitionDTO toPartition(InternalPartition topic); BrokerDTO toBrokerDto(InternalBroker broker); diff --git a/api/src/main/java/io/kafbat/ui/model/InternalTopic.java b/api/src/main/java/io/kafbat/ui/model/InternalTopic.java index 5dfab7c42..221c854f3 100644 --- a/api/src/main/java/io/kafbat/ui/model/InternalTopic.java +++ b/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; + } } diff --git a/contract-typespec/api/topics.tsp b/contract-typespec/api/topics.tsp index 7c03856bc..5fad04813 100644 --- a/contract-typespec/api/topics.tsp +++ b/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[]; } diff --git a/contract/src/main/resources/swagger/kafbat-ui-api.yaml b/contract/src/main/resources/swagger/kafbat-ui-api.yaml index 09f1c45f8..a05c9a5a4 100644 --- a/contract/src/main/resources/swagger/kafbat-ui-api.yaml +++ b/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