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

xijiu · 2025-06-29T04:55:05Z

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
Add integration test, create a cluster with 3 brokers, then shutdown
a broker and observe whether the output meets the expectations

chia7712 · 2025-06-29T06:51:18Z

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

+            // shutdown the target broker
+            int noneLeaderPartition = 2;
+            int shutdownBrokerId = clusterInstance.getLeaderBrokerId(new TopicPartition(topic, noneLeaderPartition));
+            clusterInstance.shutdownBroker(shutdownBrokerId);


Shutting down the broker could cause some admin APIs to hang, so could you please add timeout to this test in order to avoid impacting CI?

m1a2st

Thanks @xijiu for this patch, left some comments, and some tests are fail, PTAL

m1a2st · 2025-06-29T07:18:08Z

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

+        }
+
+        // append the command
+        List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics"));


Suggested change

List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics"));

List<String> cgcArgs = List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics");

m1a2st · 2025-06-29T08:21:11Z

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

+
+        private Set<TopicPartition> filterNoneLeaderPartitions(List<TopicPartition> topicPartitions) {
+            // collect all topics
+            Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());


I don't think we need to convert it to a set here, since AdminClient.describeTopics doesn’t handle duplicate topic names anyway

kafka/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

Line 2320 in a5a54dc

for (String topicName : topicNames) {

Since we still need to map to TopicPartition::topic, the current approach is acceptable.

xijiu · 2025-06-29T11:43:04Z

@chia7712 @m1a2st Thanks for CR, and I have fixed them including the failed tests, PTAL

FrankYang0529 · 2025-06-29T15:41:37Z

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

+                        && 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 a data row and no error in describe groups when a broker shutdown.");


IIUC, the topic has 3 partitions, so there're 3 data rows.

@FrankYang0529 Yeah, I have fixed it, PTAL

chia7712

@xijiu thanks for this patch. two comments are left. PTAL

chia7712 · 2025-07-07T18:20:48Z

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

@@ -592,8 +592,8 @@ 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());


excuse me, why do we need those changes?

chia7712 · 2025-07-07T18:22:50Z

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

+            Set<TopicPartition> nonLeaderTopicPartitions = filterNoneLeaderPartitions(topicPartitions);
+
+            // prepare data for partitions with leaders
+            topicPartitions.removeAll(nonLeaderTopicPartitions);


we don't need to modify the input topicPartitions if filterNoneLeaderPartitions returns the partitions having leaders.

xijiu · 2025-07-08T06:31:41Z

@chia7712 Thanks for CR, I have fixed them, PTAL

chia7712 · 2025-07-12T16:41:49Z

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

+                        .flatMap(entry -> entry.getValue().partitions().stream()
+                                .filter(partitionInfo -> partitionInfo.leader() == null)
+                                .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
+                        .collect(Collectors.toList());


.toList();

chia7712 · 2025-07-12T16:44:50Z

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

@@ -619,7 +618,12 @@ private Collection<PartitionAssignmentState> describePartitions(
                    consumerIdOpt, hostOpt, clientIdOpt, logEndOffsetOpt, leaderEpoch);
            };

-            return offsetsUtils.getLogEndOffsets(topicPartitions).entrySet().stream().map(logEndOffsetResult -> {
+            List<TopicPartition> topicPartitionsWithLeader = new ArrayList<>(topicPartitions);


List<TopicPartition> topicPartitionsWithoutLeader = filterNoneLeaderPartitions(topicPartitions); List<TopicPartition> topicPartitionsWithLeader = topicPartitions.stream().filter(tp -> !topicPartitionsWithoutLeader.contains(tp)).toList();

chia7712 · 2025-07-12T16:45:48Z

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

+            return Stream.concat(existLeaderAssignments.stream(), noneLeaderAssignments.stream())
+                    .sorted(Comparator.<PartitionAssignmentState, String>comparing(
+                            state -> state.topic.orElse(""), String::compareTo)
+                            .thenComparingInt(state -> state.partition.orElse(-1)))


the sort policy should include topic name, right?

Yeah, the original logical ordering is invalid, and I added it

Should we add test for this case?

xijiu · 2025-07-14T03:23:11Z

@chia7712 I have fixed them, PTAL ☺

TaiJuWu

Thanks for this patch, leave some comments.

TaiJuWu · 2025-07-14T03:50:10Z

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

@@ -592,8 +592,7 @@ private Collection<PartitionAssignmentState> collectConsumerAssignment(
                        getLag(Optional.empty(), Optional.empty()), consumerIdOpt, hostOpt, clientIdOpt, Optional.empty(), Optional.empty())


LOGGER is not used at Line 85 in this file, should we remove it?

Yeah, it's useless indeed, I have removed it, PTAL

TaiJuWu · 2025-07-14T04:19:22Z

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

+            // 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)


Just out of curiosity, topic is allowed to be Optional.empty or empty string ?

It maybe not allowed, it will receive a exception org.apache.kafka.common.errors.InvalidTopicException: The given topic name '' cannot be represented in a request when set the empty to a topic name

// org.apache.kafka.clients.admin.KafkaAdminClient#topicNameIsUnrepresentable

private static boolean topicNameIsUnrepresentable(String topicName) { return topicName == null || topicName.isEmpty(); }

finish coding

6e34524

github-actions bot added the triage PRs from the community label Jun 29, 2025

xijiu requested a review from FrankYang0529 June 29, 2025 04:55

github-actions bot added tools small Small PRs labels Jun 29, 2025

xijiu added ci-approved and removed small Small PRs labels Jun 29, 2025

xijiu changed the title ~~KAFKA-10357: Optimize kafka-consumer-groups.sh to return the offset info of other partitions even when the leader of some partitions are missing~~ KAFKA-10357: Optimize kafka-consumer-groups.sh to return the offset info when some partitions without leaders Jun 29, 2025

xijiu changed the title ~~KAFKA-10357: Optimize kafka-consumer-groups.sh to return the offset info when some partitions without leaders~~ KAFKA-19435: Optimize kafka-consumer-groups.sh to return the offset info when some partitions without leaders Jun 29, 2025

chia7712 reviewed Jun 29, 2025

View reviewed changes

m1a2st reviewed Jun 29, 2025

View reviewed changes

fix

b44e38e

FrankYang0529 reviewed Jun 29, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Jun 30, 2025

fix the condition details

1f8c8cd

chia7712 reviewed Jul 7, 2025

View reviewed changes

optimize and fix

3a1339c

chia7712 reviewed Jul 12, 2025

View reviewed changes

optimizion

b920624

TaiJuWu reviewed Jul 14, 2025

View reviewed changes

xijiu added 2 commits July 14, 2025 15:57

remove useless log

eeffd84

remove useless log

3193cd8

	List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics"));
	List<String> cgcArgs = List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics");

		@@ -592,8 +592,7 @@ private Collection<PartitionAssignmentState> collectConsumerAssignment(
		getLag(Optional.empty(), Optional.empty()), consumerIdOpt, hostOpt, clientIdOpt, Optional.empty(), Optional.empty())

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

Are you sure you want to change the base?

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

Conversation

xijiu commented Jun 29, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

m1a2st left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xijiu commented Jun 29, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xijiu commented Jul 8, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xijiu commented Jul 14, 2025

Uh oh!

TaiJuWu left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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

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

xijiu commented Jun 29, 2025 •

edited by github-actions bot

Loading

m1a2st left a comment •

edited

Loading