Repository: provectus/kafka-ui
Issue: Sort consumer groups by messages behind
Labels: type/enhancement
good first issue
scope/backend
scope/frontend
status/accepted

### Discussed in https://github.com/provectus/kafka-ui/discussions/3467

<div type='discussions-op-text'>

<sup>Originally posted by **yardenshoham** March  9, 2023</sup>
In the UI we can sort by `Group ID`, `Num Of Members` or `State` but not `Num of Topics`, `Messages Behind` nor `Coordinator`.
![image](https://user-images.githubusercontent.com/20454870/224008985-f275e71d-44b6-4234-bb58-e6400ce6d719.png)
</div>

I suppose one would need to add logic here? https://github.com/provectus/kafka-ui/blob/334ba3df99dfc84385faace167f6410c8ce0be91/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java#L134-L160

Issue comments:
@Haarolean can you hint at how this could be achieved? I want to try
I opened a PR with @nisanohana3, #3527 
@yardenshoham yep, I've seen it, thanks! Just assigning the issue

Pull request review comments:
please rename method to `getMessagesBehind(..)` or `calculateMessagesBehind(..)`
not needed here, please remove
I suggest either `MESSAGES_BEHIND` or `CONSUMER_LAG`
Done
no need to add description into InternalConsumerGroup
here how it can be done
```
      case MESSAGES_BEHIND -> {
        record GroupWithDescr(InternalConsumerGroup g, ConsumerGroupDescription d) { }

        Comparator<GroupWithDescr> comparator = Comparator.comparingLong(gwd ->
            gwd.g.getMessagesBehind() == null ? 0L : gwd.g.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(g -> Streams.zip(g.stream(), descriptions.stream(), GroupWithDescr::new).toList())
                      .map(g -> sortAndPaginate(g, comparator, pageNum, perPage, sortOrderDto)
                          .map(GroupWithDescr::d).toList());
                }
            );
      }
```
Done
Done
Done

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ConsumerGroupMapper.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ConsumerGroupMapper.java
index 9f7e32ed148..21d9efda9ca 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ConsumerGroupMapper.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ConsumerGroupMapper.java
@@ -89,19 +89,7 @@ private static <T extends ConsumerGroupDTO> T convertToConsumerGroup(
             .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());
 
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalConsumerGroup.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalConsumerGroup.java
index d7b3a732f1f..e8199fa8ef5 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalConsumerGroup.java
+++ b/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 static InternalConsumerGroup create(
     );
     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;
+  }
+
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java
index 024eb3df513..e848146881d 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java
+++ b/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 @@ private Mono<List<ConsumerGroupDescription>> loadSortedDescriptions(ReactiveAdmi
             .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());
+                }
+            );
+      }
     };
   }
 
diff --git a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
index ea335f282cf..fe9723b2dd3 100644
--- a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
+++ b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
@@ -2416,6 +2416,7 @@ components:
         - NAME
         - MEMBERS
         - STATE
+        - MESSAGES_BEHIND
 
     ConsumerGroupsPageResponse:
       type: object
diff --git a/kafka-ui-react-app/src/components/ConsumerGroups/List.tsx b/kafka-ui-react-app/src/components/ConsumerGroups/List.tsx
index feb772ac826..ef8d73f5156 100644
--- a/kafka-ui-react-app/src/components/ConsumerGroups/List.tsx
+++ b/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',


Test patch: