Allow sorting consumer groups by messages behind (#3527)

* Allow sorting consumer groups by messages behind
* Added an additional value through the enum of ConsumerGroupOrdering
* Enabled sorting in the react consumer groups page (by messages behind)
* Moved the message behind calculation logic to InternalConsumerGroup as part of its creation
* Added the messages behind case - sorting internalConsumerGroup according to comparator with messages behind logic

Co-authored-by: Yarden Shoham <hrsi88@gmail.com>
Signed-off-by: nisanohana3 <nisana230@gmail.com>

kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ConsumerGroupMapper.java

@@ -89,19 +89,7 @@ public class ConsumerGroupMapper {
             .flatMap(m -> m.getAssignment().stream().map(TopicPartition::topic))
     ).collect(Collectors.toSet()).size();
 
-    Long messagesBehind = null;
-    // messagesBehind should be undefined if no committed offsets found for topic
-    if (!c.getOffsets().isEmpty()) {
-      messagesBehind = c.getOffsets().entrySet().stream()
-          .mapToLong(e ->
-              Optional.ofNullable(c.getEndOffsets())
-                  .map(o -> o.get(e.getKey()))
-                  .map(o -> o - e.getValue())
-                  .orElse(0L)
-          ).sum();
-    }
-
-    consumerGroup.setMessagesBehind(messagesBehind);
+    consumerGroup.setMessagesBehind(c.getMessagesBehind());
     consumerGroup.setTopics(numTopics);
     consumerGroup.setSimple(c.isSimple());
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalConsumerGroup.java

@@ -20,6 +20,7 @@ public class InternalConsumerGroup {
   private final Collection<InternalMember> members;
   private final Map<TopicPartition, Long> offsets;
   private final Map<TopicPartition, Long> endOffsets;
+  private final Long messagesBehind;
   private final String partitionAssignor;
   private final ConsumerGroupState state;
   private final Node coordinator;
@@ -58,7 +59,25 @@ public class InternalConsumerGroup {
     );
     builder.offsets(groupOffsets);
     builder.endOffsets(topicEndOffsets);
+    builder.messagesBehind(calculateMessagesBehind(groupOffsets, topicEndOffsets));
     Optional.ofNullable(description.coordinator()).ifPresent(builder::coordinator);
     return builder.build();
   }
+
+  private static Long calculateMessagesBehind(Map<TopicPartition, Long> offsets, Map<TopicPartition, Long> endOffsets) {
+    Long messagesBehind = null;
+    // messagesBehind should be undefined if no committed offsets found for topic
+    if (!offsets.isEmpty()) {
+      messagesBehind = offsets.entrySet().stream()
+          .mapToLong(e ->
+              Optional.ofNullable(endOffsets)
+                  .map(o -> o.get(e.getKey()))
+                  .map(o -> o - e.getValue())
+                  .orElse(0L)
+          ).sum();
+    }
+
+    return messagesBehind;
+  }
+
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.service;
 
+import com.google.common.collect.Streams;
 import com.google.common.collect.Table;
 import com.provectus.kafka.ui.model.ConsumerGroupOrderingDTO;
 import com.provectus.kafka.ui.model.InternalConsumerGroup;
@@ -157,6 +158,24 @@ public class ConsumerGroupService {
             .map(descriptions ->
                 sortAndPaginate(descriptions.values(), comparator, pageNum, perPage, sortOrderDto).toList());
       }
+      case MESSAGES_BEHIND -> {
+        record GroupWithDescr(InternalConsumerGroup icg, ConsumerGroupDescription cgd) { }
+
+        Comparator<GroupWithDescr> comparator = Comparator.comparingLong(gwd ->
+            gwd.icg.getMessagesBehind() == null ? 0L : gwd.icg.getMessagesBehind());
+
+        var groupNames = groups.stream().map(ConsumerGroupListing::groupId).toList();
+
+        yield ac.describeConsumerGroups(groupNames)
+            .flatMap(descriptionsMap -> {
+                  List<ConsumerGroupDescription> descriptions = descriptionsMap.values().stream().toList();
+                  return getConsumerGroups(ac, descriptions)
+                      .map(icg -> Streams.zip(icg.stream(), descriptions.stream(), GroupWithDescr::new).toList())
+                      .map(gwd -> sortAndPaginate(gwd, comparator, pageNum, perPage, sortOrderDto)
+                            .map(GroupWithDescr::cgd).toList());
+                }
+            );
+      }
     };
   }
 

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

@@ -2416,6 +2416,7 @@ components:
         - NAME
         - MEMBERS
         - STATE
+        - MESSAGES_BEHIND
 
     ConsumerGroupsPageResponse:
       type: object

kafka-ui-react-app/src/components/ConsumerGroups/List.tsx

@@ -56,9 +56,9 @@ const List = () => {
         enableSorting: false,
       },
       {
+        id: ConsumerGroupOrdering.MESSAGES_BEHIND,
         header: 'Messages Behind',
         accessorKey: 'messagesBehind',
-        enableSorting: false,
       },
       {
         header: 'Coordinator',