Repository: provectus/kafka-ui
Issue: Consumers: Add sorting by additional columns
Labels: type/enhancement
good first issue
scope/backend
scope/frontend
status/accepted

<!--

Don't forget to check for existing issues/discussions regarding your proposal. We might already have it.
https://github.com/provectus/kafka-ui/issues
https://github.com/provectus/kafka-ui/discussions

-->

**Describe the bug**
<!--(A clear and concise description of what the bug is.)-->
Sorting only possible by the Num of messages and State. Not possible by the Num of topics or Messages behind

**Expected behavior**
<!--
(A clear and concise description of what you expected to happen)
-->
Sorting by the "Num of topics" and "Messages behind" should be possible as well.

**Set up**
<!--
How do you run the app? Please provide as much info as possible:
1. App version (docker image version or check commit hash in the top left corner in UI)
2. Helm chart version, if you use one
3. Any IAAC configs

We might close the issue without further explanation if you don't provide such information.
-->
[f4e6afe](https://github.com/provectus/kafka-ui/commit/f4e6afe)

**Steps to Reproduce**
<!-- We'd like you to provide an example setup (via docker-compose, helm, etc.) 
to reproduce the problem, especially with a complex setups. -->
Steps to reproduce the behavior:

1. Log in to Kafka UI and navigate to the Consumers page.

**Screenshots**
<!--
(If applicable, add screenshots to help explain your problem)
-->
![image](https://user-images.githubusercontent.com/112083452/209812552-3e868093-2a51-4d59-9e67-7680b07d7fc1.png)


**Additional context**
<!--
(Add any other context about the problem here)
-->
@Haarolean requested to create a separate issue from the #2651 "Please raise an issue for sorting by num of topics and messages behind."

Issue comments:
Sorting by "Num of Topics" and by "messages Behind" was added and merged to the master. 
I am adding references to the relevant PRs:
[https://github.com/provectus/kafka-ui/pull/3633](url)
[https://github.com/provectus/kafka-ui/pull/3527](url)

I think this issue can be closed


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: