You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "xijiu (via GitHub)" <gi...@apache.org> on 2025/07/15 06:36:05 UTC
[PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
xijiu opened a new pull request, #20168:
URL: https://github.com/apache/kafka/pull/20168
1. Add check leader missing logic in method `ConsumerGroupCommand.ConsumerGroupService#prepareOffsetsToReset` in order to fail quickly
2. Add some tests
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2227295797
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
+ }
+
+ // check the partitions have leader
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
Review Comment:
```
chia7712@chia7712-ubuntu:~/project/kafka$ ./bin/kafka-consumer-groups.sh \
--bootstrap-server 172.20.10.2:20001 \
--reset-offsets \
--to-earliest \
--execute \
--group perf-consumer-19460 \
--topic chia:1
Error: Executing consumer group command failed due to The partitions "chia-2" have no leader
org.apache.kafka.common.errors.LeaderNotAvailableException: The partitions "chia-2" have no leader
```
It is indeed a bug that unrelated topic partitions could fail the tool.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2228677312
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java:
##########
@@ -659,6 +662,41 @@ public void testResetWithUnrecognizedNewConsumerOption(ClusterInstance cluster)
assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
}
+ @ClusterTest(brokers = 3, serverProperties = {@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")})
+ public void testResetOffsetsWithPartitionNoneLeader(ClusterInstance cluster) throws Exception {
+ String group = generateRandomGroupId();
+ String topic = generateRandomTopic();
+ String[] args = buildArgsForGroup(cluster, group, "--topic", topic + ":0,1,2",
+ "--to-earliest", "--execute");
+
+ try (Admin admin = cluster.admin();
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
+
+ admin.createTopics(singleton(new NewTopic(topic, 3, (short) 1))).all().get();
+ produceConsumeAndShutdown(cluster, topic, group, 2, GroupProtocol.CLASSIC);
+ assertDoesNotThrow(() -> resetOffsets(service));
+ // shutdown a broker to make some partitions missing leader
+ cluster.shutdownBroker(0);
Review Comment:
The shutdown already has a 5-minute timeout by default.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "Yunyung (via GitHub)" <gi...@apache.org>.
Yunyung commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2228628205
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java:
##########
@@ -659,6 +662,41 @@ public void testResetWithUnrecognizedNewConsumerOption(ClusterInstance cluster)
assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
}
+ @ClusterTest(brokers = 3, serverProperties = {@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")})
+ public void testResetOffsetsWithPartitionNoneLeader(ClusterInstance cluster) throws Exception {
+ String group = generateRandomGroupId();
+ String topic = generateRandomTopic();
+ String[] args = buildArgsForGroup(cluster, group, "--topic", topic + ":0,1,2",
+ "--to-earliest", "--execute");
+
+ try (Admin admin = cluster.admin();
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
+
+ admin.createTopics(singleton(new NewTopic(topic, 3, (short) 1))).all().get();
+ produceConsumeAndShutdown(cluster, topic, group, 2, GroupProtocol.CLASSIC);
+ assertDoesNotThrow(() -> resetOffsets(service));
+ // shutdown a broker to make some partitions missing leader
+ cluster.shutdownBroker(0);
Review Comment:
Should we wait to ensure it has shut down correctly?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2207591681
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java:
##########
@@ -659,6 +663,26 @@ public void testResetWithUnrecognizedNewConsumerOption(ClusterInstance cluster)
assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
}
+ @Timeout(60)
Review Comment:
Nit: No need to add `@Timeout` here, `@ClusterTest` already includes a timeout setting.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2227735419
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
+ }
+
+ // check the partitions have leader
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
Review Comment:
Yes, that's a very good point. Thanks for finding it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2227310093
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
+ }
+
+ // check the partitions have leader
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
Review Comment:
`filterNoneLeaderPartitions` needs to get fixed since it could return unrelated topic partitions.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on PR #20168:
URL: https://github.com/apache/kafka/pull/20168#issuecomment-3082860443
> I think this could do with a little more work. If I create a topic `T1` with 1 partition, the following command `bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group CG1 --to-earliest --topic T1:1 --execute` still hangs for a long time, even though it has called `Admin.describeTopics`.
>
> I think it can be improved for the case of a non-existent topic of an existing topic. Seems like it's worth fixing this at the same time.
Good find! It's indeed hangs when a non-existent TopicPartition of an existing topic.
I have added some logic to determine whether the target TopicPartitions actually exists, PTAL
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2209031995
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1028,21 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
+ if (!partitionsWithoutLeader.isEmpty()) {
+ // append the TopicPartition list string
Review Comment:
@AndrewJSchofield Thanks very much for CR, it's indeed much neater. I have fixed it, PTAL
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #20168:
URL: https://github.com/apache/kafka/pull/20168
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2221826208
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
+ }
+
+ // check the partitions have leader
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
Review Comment:
if there is a topic having three partitions, and the `t-2` partition is offline, then if `partitionsToReset` is `t-0,t-1`, `filterNoneLeaderPartitions` will return `t-2`, causing the tool to fail. Is it expected?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on PR #20168:
URL: https://github.com/apache/kafka/pull/20168#issuecomment-3072209689
The all methods which depend on `Admin#listOffsets` are as follows:
- offsetsUtils.resetToOffset
- offsetsUtils.resetToEarliest
- offsetsUtils.resetToLatest
- offsetsUtils.resetByShiftBy
- offsetsUtils.resetToDateTime
- offsetsUtils.resetByDuration
- offsetsUtils.resetFromFile
- offsetsUtils.resetToCurrent
And they are all called in method `ConsumerGroupCommand.ConsumerGroupService#prepareOffsetsToReset`, so we should add the check logic in this method.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #20168:
URL: https://github.com/apache/kafka/pull/20168#issuecomment-3088631223
Also please merge latest changes from trunk. Hopefully you'll get a clean test run then.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2207380826
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1028,21 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
+ if (!partitionsWithoutLeader.isEmpty()) {
+ // append the TopicPartition list string
Review Comment:
nit: You could do this string concatenation much more neatly with something like
```
String partitionStr = partitionsWithoutLeader.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2227748431
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
+ }
+
+ // check the partitions have leader
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
Review Comment:
file https://github.com/apache/kafka/pull/20235 to fix it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2209034003
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java:
##########
@@ -659,6 +663,26 @@ public void testResetWithUnrecognizedNewConsumerOption(ClusterInstance cluster)
assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
}
+ @Timeout(60)
Review Comment:
@m1a2st Thanks very much, the `@Timeout` is redundant, and I have remove it, PTAL
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "xijiu (via GitHub)" <gi...@apache.org>.
xijiu commented on PR #20168:
URL: https://github.com/apache/kafka/pull/20168#issuecomment-3090143092
@AndrewJSchofield Thanks for the code review, and I have fixed them and merged the latest changes from trunk, PTAL
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2222305652
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1024,6 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNonExistentPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" do not exist");
+ }
+
+ // check the partitions have leader
+ List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
Review Comment:
I think it's fair enough.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20168:
URL: https://github.com/apache/kafka/pull/20168#discussion_r2215468267
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1028,15 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
- private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNotExistPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" does not exist");
+ }
+
+ // check the partitions have leader
List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
if (!partitionsWithoutLeader.isEmpty()) {
String partitionStr = partitionsWithoutLeader.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
- // throw exception
throw new LeaderNotAvailableException("The partitions \"" + partitionStr + "\" have no leader");
}
}
+ private List<TopicPartition> filterNotExistPartitions(Collection<TopicPartition> topicPartitions) {
+ // collect all topics
+ Set<String> topics = topicPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
+ try {
+ List<TopicPartition> existPartitions = adminClient.describeTopics(topics).allTopicNames().get().entrySet()
+ .stream()
+ .flatMap(entry -> entry.getValue().partitions().stream()
+ .map(partitionInfo -> new TopicPartition(entry.getKey(), partitionInfo.partition())))
+ .toList();
+
+ return topicPartitions.stream().filter(element -> !existPartitions.contains(element)).toList();
+ } catch (Exception e) {
Review Comment:
For consistency, we tend to ` catch (InterruptedException | ExecutionException e)` in this source file.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1028,15 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
- private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNotExistPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" does not exist");
Review Comment:
nit: Grammar "do not exist" I think.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -1028,15 +1029,38 @@ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String grou
return null;
}
- private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
+ private void checkAllTopicPartitionsValid(Collection<TopicPartition> partitionsToReset) {
+ // check the partitions exist
+ List<TopicPartition> partitionsNotExistList = filterNotExistPartitions(partitionsToReset);
+ if (!partitionsNotExistList.isEmpty()) {
+ String partitionStr = partitionsNotExistList.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
+ throw new UnknownTopicOrPartitionException("The partitions \"" + partitionStr + "\" does not exist");
+ }
+
+ // check the partitions have leader
List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
if (!partitionsWithoutLeader.isEmpty()) {
String partitionStr = partitionsWithoutLeader.stream().map(TopicPartition::toString).collect(Collectors.joining(","));
- // throw exception
throw new LeaderNotAvailableException("The partitions \"" + partitionStr + "\" have no leader");
}
}
+ private List<TopicPartition> filterNotExistPartitions(Collection<TopicPartition> topicPartitions) {
Review Comment:
nit: `filterNonExistentPartitions` is probably a bit better.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org