muralibasani commented on code in PR #24098:
URL: https://github.com/apache/kafka/pull/24098#discussion_r4496850537
##########
tools/src/main/java/org/apache/kafka/tools/GroupOffsetsResetter.java:
##########
@@ -459,17 +459,22 @@ public List<TopicPartition>
filterNoneLeaderPartitions(Collection<TopicPartition
// collect all topics
Set<String> topics =
topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
- try {
- return adminClient.describeTopics(topics, withTimeoutMs(new
DescribeTopicsOptions())).allTopicNames().get().entrySet()
- .stream()
- .flatMap(entry -> entry.getValue().partitions().stream()
+ List<TopicPartition> noneLeaderPartitions = new ArrayList<>();
+ adminClient.describeTopics(topics, withTimeoutMs(new
DescribeTopicsOptions())).topicNameValues().forEach((topic, future) -> {
+ try {
+ future.get().partitions().stream()
.filter(partitionInfo -> partitionInfo.leader() == null)
- .map(partitionInfo -> new TopicPartition(entry.getKey(),
partitionInfo.partition())))
+ .map(partitionInfo -> new TopicPartition(topic,
partitionInfo.partition()))
.filter(topicPartitions::contains)
Review Comment:
Wondering how it would behave when a topic is deleted and recreated with
fewer partitions.
Group can still be associated to old partitions in an assignment?
In case there are left over partitions (partition that's gone from a topic
that still exists) as they don't have metadata and would lead to any
retries/fail/timeout ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
