KAFKA-19435 Optimize kafka-consumer-groups.sh to return the offset info when some partitions without leaders (apache#20064)

1. Optimize the corresponding logic in the `ConsumerGroupCommand` by
first checking if a leader exists for the partition before invoking the
`admin.listOffsets`. Finally, concatenate the data and return
2. Add integration test, create a cluster with 3 brokers, then shutdown
a broker and observe whether the output meets the expectations

Reviewers: Ken Huang <s7133700@gmail.com>, PoAn Yang
 <payang@apache.org>, TaiJuWu <tjwu1217@gmail.com>, Chia-Ping Tsai
 <chia7712@gmail.com>

Author: xijiu

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

@@ -51,9 +51,6 @@
 import com.google.re2j.Pattern;
 import com.google.re2j.PatternSyntaxException;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.ArrayList;
@@ -82,7 +79,6 @@
 import joptsimple.OptionSpec;
 
 public class ConsumerGroupCommand {
-    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerGroupCommand.class);
 
     static final String MISSING_COLUMN_VALUE = "-";
 
@@ -592,8 +588,7 @@ private Collection<PartitionAssignmentState> collectConsumerAssignment(
                         getLag(Optional.empty(), Optional.empty()), consumerIdOpt, hostOpt, clientIdOpt, Optional.empty(), Optional.empty())
                 );
             } else {
-                List<TopicPartition> topicPartitionsSorted = topicPartitions.stream().sorted(Comparator.comparingInt(TopicPartition::partition)).collect(Collectors.toList());
-                return describePartitions(group, coordinator, topicPartitionsSorted, committedOffsets, consumerIdOpt, hostOpt, clientIdOpt);
+                return describePartitions(group, coordinator, topicPartitions, committedOffsets, consumerIdOpt, hostOpt, clientIdOpt);
             }
         }
 
@@ -604,7 +599,7 @@ private Optional<Long> getLag(Optional<Long> offset, Optional<Long> logEndOffset
         private Collection<PartitionAssignmentState> describePartitions(
             String group,
             Optional<Node> coordinator,
-            List<TopicPartition> topicPartitions,
+            Collection<TopicPartition> topicPartitions,
             Map<TopicPartition, OffsetAndMetadata> committedOffsets,
             Optional<String> consumerIdOpt,
             Optional<String> hostOpt,
@@ -619,7 +614,11 @@ private Collection<PartitionAssignmentState> describePartitions(
                     consumerIdOpt, hostOpt, clientIdOpt, logEndOffsetOpt, leaderEpoch);
             };
 
-            return offsetsUtils.getLogEndOffsets(topicPartitions).entrySet().stream().map(logEndOffsetResult -> {
+            List<TopicPartition> topicPartitionsWithoutLeader = filterNoneLeaderPartitions(topicPartitions);
+            List<TopicPartition> topicPartitionsWithLeader = topicPartitions.stream().filter(tp -> !topicPartitionsWithoutLeader.contains(tp)).toList();
+
+            // prepare data for partitions with leaders
+            List<PartitionAssignmentState> existLeaderAssignments = offsetsUtils.getLogEndOffsets(topicPartitionsWithLeader).entrySet().stream().map(logEndOffsetResult -> {
                 if (logEndOffsetResult.getValue() instanceof OffsetsUtils.LogOffset)
                     return getDescribePartitionResult.apply(
                         logEndOffsetResult.getKey(),
@@ -631,7 +630,34 @@ else if (logEndOffsetResult.getValue() instanceof OffsetsUtils.Ignore)
                     return null;
 
                 throw new IllegalStateException("Unknown LogOffset subclass: " + logEndOffsetResult.getValue());
-            }).collect(Collectors.toList());
+            }).toList();
+
+            // prepare data for partitions without leaders
+            List<PartitionAssignmentState> noneLeaderAssignments = topicPartitionsWithoutLeader.stream()
+                    .map(tp -> getDescribePartitionResult.apply(tp, Optional.empty())).toList();
+
+            // concat the data and then sort them
+            return Stream.concat(existLeaderAssignments.stream(), noneLeaderAssignments.stream())
+                    .sorted(Comparator.<PartitionAssignmentState, String>comparing(
+                            state -> state.topic.orElse(""), String::compareTo)
+                            .thenComparingInt(state -> state.partition.orElse(-1)))
+                    .collect(Collectors.toList());
+        }
+
+        private List<TopicPartition> filterNoneLeaderPartitions(Collection<TopicPartition> topicPartitions) {
+            // collect all topics
+            Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
+
+            try {
+                return adminClient.describeTopics(topics).allTopicNames().get().entrySet()
+                        .stream()
+                        .flatMap(entry -> entry.getValue().partitions().stream()
+                                .filter(partitionInfo -> partitionInfo.leader() == null)
+                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
+                        .toList();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
         }
 
         Map<String, Map<TopicPartition, OffsetAndMetadata>> resetOffsets() {

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

@@ -91,6 +91,8 @@ public void testAdminRequestsForDescribeOffsets() throws Exception {
                 .thenReturn(listGroupOffsetsResult(GROUP));
         when(admin.listOffsets(offsetsArgMatcher(), any()))
                 .thenReturn(listOffsetsResult());
+        when(admin.describeTopics(ArgumentMatchers.anySet()))
+                .thenReturn(describeTopicsResult());
 
         Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> statesAndAssignments = groupService.collectGroupOffsets(GROUP);
         assertEquals(Optional.of(GroupState.STABLE), statesAndAssignments.getKey());
@@ -174,6 +176,7 @@ public void testAdminRequestsForDescribeNegativeOffsets() throws Exception {
                 any()
         )).thenReturn(new ListOffsetsResult(endOffsets.entrySet().stream().filter(e -> unassignedTopicPartitions.contains(e.getKey()))
                 .collect(Collectors.toMap(Entry::getKey, Entry::getValue))));
+        when(admin.describeTopics(ArgumentMatchers.anySet())).thenReturn(describeTopicsResult());
 
         Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> statesAndAssignments = groupService.collectGroupOffsets(GROUP);
         Optional<GroupState> state = statesAndAssignments.getKey();
@@ -289,6 +292,18 @@ private Map<TopicPartition, OffsetSpec> offsetsArgMatcher() {
         );
     }
 
+    private DescribeTopicsResult describeTopicsResult() {
+        Map<String, TopicDescription> topicDescriptionMap = TOPICS.stream().collect(Collectors.toMap(
+                Function.identity(),
+                topic -> new TopicDescription(
+                        topic,
+                        false,
+                        IntStream.range(0, NUM_PARTITIONS)
+                                .mapToObj(i -> new TopicPartitionInfo(i, Node.noNode(), List.of(), List.of()))
+                                .toList())));
+        return AdminClientTestUtils.describeTopicsResult(topicDescriptionMap);
+    }
+
     private ListOffsetsResult listOffsetsResult() {
         ListOffsetsResultInfo resultInfo = new ListOffsetsResultInfo(100, System.currentTimeMillis(), Optional.of(1));
         Map<TopicPartition, KafkaFuture<ListOffsetsResultInfo>> futures = TOPIC_PARTITIONS.stream().collect(Collectors.toMap(

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

@@ -47,6 +47,7 @@
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -59,6 +60,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
@@ -1053,6 +1055,64 @@ public void testDescribeNonOffsetCommitGroup(ClusterInstance clusterInstance) th
         }
     }
 
+    /**
+     * The config `OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG` needs to be set to a value greater than 1 to ensure the
+     * normal invocation of APIs such as `FIND_COORDINATOR` when a broker has shutdown
+     */
+    @Timeout(60)
+    @ClusterTest(brokers = 3, serverProperties = {@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")})
+    public void testDescribeConsumerGroupWithoutLeaders(ClusterInstance clusterInstance) throws Exception {
+        int brokerNum = 3;
+        this.clusterInstance = clusterInstance;
+
+        // define topic and group, then send 5 records to each partition
+        String topic = TOPIC_PREFIX + UUID.randomUUID();
+        String group = GROUP_PREFIX + UUID.randomUUID();
+        clusterInstance.createTopic(topic, brokerNum, (short) 1);
+        for (int i = 0; i < brokerNum; i++) {
+            sendRecords(topic, i, 5);
+        }
+
+        // append the command
+        List<String> cgcArgs = List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics");
+
+        try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(GroupProtocol.CLASSIC, group, topic, Map.of());
+             ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]));
+             Admin admin = clusterInstance.admin()
+        ) {
+            // shutdown the target broker
+            int noneLeaderPartition = 2;
+            int shutdownBrokerId = clusterInstance.getLeaderBrokerId(new TopicPartition(topic, noneLeaderPartition));
+            clusterInstance.shutdownBroker(shutdownBrokerId);
+
+            TestUtils.waitForCondition(() -> {
+                Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
+                String[] lines = res.getKey().trim().split("\n");
+                if (lines.length != 4 || !res.getValue().isEmpty()) {
+                    return false;
+                }
+
+                // get the client data, such as `consumerId,host,clientId`, to append the expected output
+                ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();
+                MemberDescription memberDescription = consumerGroupDescription.members().iterator().next();
+                String consumerId = memberDescription.consumerId();
+                String host = memberDescription.host();
+                String clientId = memberDescription.clientId();
+
+                // the expected output
+                List<String> partition0content = List.of(group, topic, "0", "5", "5", "0", consumerId, host, clientId);
+                List<String> partition1content = List.of(group, topic, "1", "5", "5", "0", consumerId, host, clientId);
+                List<String> partition2content = List.of(group, topic, "2", "-", "-", "-", consumerId, host, clientId);
+
+                return checkArgsHeaderOutput(cgcArgs, lines[0])
+                        && Arrays.stream(lines[1].trim().split("\\s+")).toList().equals(partition0content)
+                        && Arrays.stream(lines[2].trim().split("\\s+")).toList().equals(partition1content)
+                        && Arrays.stream(lines[3].trim().split("\\s+")).toList().equals(partition2content);
+            }, "Expected 3 data rows excluding the header and no error in describe groups when a broker shutdown.");
+        }
+    }
+
+
     @Test
     public void testDescribeWithUnrecognizedNewConsumerOption() {
         String group = GROUP_PREFIX +  "unrecognized";