KAFKA-19944: kafka-share-groups.sh describe offsets message improvement (#21024)

kafka-share-groups.sh --describe --offsets prints an empty table
consisting only of headers when there is no offset information to
display for a share group. This is most likely only when a topic which
was formerly being consumed is deleted. This PR displays a message
instead of an empty table, similar to how the tool works if the list of
members is requested and there are no active members.

Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>

Author: AndrewJSchofield

tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java

@@ -545,35 +545,39 @@ private void printOffsets(TreeMap<String, Entry<ShareGroupDescription, Collectio
                         .thenComparingInt(info -> info.partition))
                     .toList();
 
-                String fmt = printOffsetFormat(groupId, offsetsInfo, verbose);
-
-                if (verbose) {
-                    System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "LEADER-EPOCH", "START-OFFSET", "LAG");
+                if (offsetsInfo.isEmpty()) {
+                    System.out.println("\nShare group '" + groupId + "' has no offset information.");
                 } else {
-                    System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "START-OFFSET", "LAG");
-                }
+                    String fmt = printOffsetFormat(groupId, offsetsInfo, verbose);
 
-                for (SharePartitionOffsetInformation info : offsetsInfo) {
                     if (verbose) {
-                        System.out.printf(fmt,
-                            groupId,
-                            info.topic,
-                            info.partition,
-                            info.leaderEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
-                            info.offset.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
-                            info.lag.map(Object::toString).orElse(MISSING_COLUMN_VALUE)
-                        );
+                        System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "LEADER-EPOCH", "START-OFFSET", "LAG");
                     } else {
-                        System.out.printf(fmt,
-                            groupId,
-                            info.topic,
-                            info.partition,
-                            info.offset.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
-                            info.lag.map(Object::toString).orElse(MISSING_COLUMN_VALUE)
-                        );
+                        System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "START-OFFSET", "LAG");
                     }
+
+                    for (SharePartitionOffsetInformation info : offsetsInfo) {
+                        if (verbose) {
+                            System.out.printf(fmt,
+                                groupId,
+                                info.topic,
+                                info.partition,
+                                info.leaderEpoch.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+                                info.offset.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+                                info.lag.map(Object::toString).orElse(MISSING_COLUMN_VALUE)
+                            );
+                        } else {
+                            System.out.printf(fmt,
+                                groupId,
+                                info.topic,
+                                info.partition,
+                                info.offset.map(Object::toString).orElse(MISSING_COLUMN_VALUE),
+                                info.lag.map(Object::toString).orElse(MISSING_COLUMN_VALUE)
+                            );
+                        }
+                    }
+                    System.out.println();
                 }
-                System.out.println();
             });
         }
 

tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java

@@ -286,6 +286,47 @@ public void testDescribeOffsetsOfExistingGroupWithNulls() throws Exception {
         }
     }
 
+    @Test
+    public void testDescribeOffsetsOfExistingGroupWithNoOffsetInfo() throws Exception {
+        String firstGroup = "group1";
+        String bootstrapServer = "localhost:9092";
+
+        for (List<String> describeType : DESCRIBE_TYPE_OFFSETS) {
+            List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", bootstrapServer, "--describe", "--group", firstGroup));
+            cgcArgs.addAll(describeType);
+            Admin adminClient = mock(KafkaAdminClient.class);
+            DescribeShareGroupsResult describeShareGroupsResult = mock(DescribeShareGroupsResult.class);
+            ShareGroupDescription exp = new ShareGroupDescription(
+                firstGroup,
+                List.of(),
+                GroupState.EMPTY,
+                new Node(0, "host1", 9090), 0, 0);
+            // When there is no offset information at all, an empty map will be returned
+            ListShareGroupOffsetsResult listShareGroupOffsetsResult = AdminClientTestUtils.createListShareGroupOffsetsResult(
+                Map.of(
+                    firstGroup,
+                    KafkaFuture.completedFuture(Map.of())
+                )
+            );
+
+            when(describeShareGroupsResult.describedGroups()).thenReturn(Map.of(firstGroup, KafkaFuture.completedFuture(exp)));
+            when(adminClient.describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class))).thenReturn(describeShareGroupsResult);
+            when(adminClient.listShareGroupOffsets(ArgumentMatchers.anyMap(), any(ListShareGroupOffsetsOptions.class))).thenReturn(listShareGroupOffsetsResult);
+            try (ShareGroupService service = getShareGroupService(cgcArgs.toArray(new String[0]), adminClient)) {
+                TestUtils.waitForCondition(() -> {
+                    Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
+                    String[] lines = res.getKey().trim().split("\n");
+                    if (lines.length != 1 && !res.getValue().isEmpty()) {
+                        return false;
+                    }
+
+                    String expectedValue = "Share group '" + firstGroup + "' has no offset information.";
+                    return expectedValue.equals(lines[0]);
+                }, "Expected just an informational message with describe type " + String.join(" ", describeType) + ".");
+            }
+        }
+    }
+
     @Test
     public void testDescribeOffsetsOfAllExistingGroups() throws Exception {
         String firstGroup = "group1";