KAFKA-19435: Optimize kafka-consumer-groups.sh to return the offset info when some partitions without leaders
#20064
+110
−4
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
xijiu
changed the title
kafka-consumer-groups.sh to return the offset info of other partitions even when the leader of some partitions are missing
xijiu
changed the title
kafka-consumer-groups.sh to return the offset info when some partitions without leaders
chia7712
reviewed
Jun 29, 2025
| // shutdown the target broker | ||
| int noneLeaderPartition = 2; | ||
| int shutdownBrokerId = clusterInstance.getLeaderBrokerId(new TopicPartition(topic, noneLeaderPartition)); | ||
| clusterInstance.shutdownBroker(shutdownBrokerId); |
There was a problem hiding this comment.
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
reviewed
Jun 29, 2025
| } | ||
|
|
||
| // append the command | ||
| List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics")); |
There was a problem hiding this comment.
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"); |
|
|
||
| private Set<TopicPartition> filterNoneLeaderPartitions(List<TopicPartition> topicPartitions) { | ||
| // collect all topics | ||
| Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet()); |
There was a problem hiding this comment.
I don't think we need to convert it to a set here, since AdminClient.describeTopics doesn’t handle duplicate topic names anyway
There was a problem hiding this comment.
Since we still need to map to TopicPartition::topic, the current approach is acceptable.
| && 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."); |
There was a problem hiding this comment.
IIUC, the topic has 3 partitions, so there're 3 data rows.
There was a problem hiding this comment.
@FrankYang0529 Yeah, I have fixed it, PTAL
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
ConsumerGroupCommandbyfirst checking if a leader exists for the partition before invoking the
admin.listOffsets. Finally, concatenate the data and returna broker and observe whether the output meets the expectations