KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable #20168

xijiu · 2025-07-15T06:36:03Z

Add check leader missing logic in method
ConsumerGroupCommand.ConsumerGroupService#prepareOffsetsToReset in
order to fail quickly
Add some tests

Reviewers: TaiJuWu [email protected], Lan Ding [email protected],
Ken Huang [email protected], Andrew Schofield
[email protected]

xijiu · 2025-07-15T06:36:34Z

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.

AndrewJSchofield

Thanks for the PR. Just one minor comment.

AndrewJSchofield · 2025-07-15T12:45:34Z

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

+        private void checkAllTopicPartitionsHaveLeader(Collection<TopicPartition> partitionsToReset) {
+            List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);
+            if (!partitionsWithoutLeader.isEmpty()) {
+                // append the TopicPartition list string


nit: You could do this string concatenation much more neatly with something like

String partitionStr = partitionsWithoutLeader.stream().map(TopicPartition::toString).collect(Collectors.joining(","));

@AndrewJSchofield Thanks very much for CR, it's indeed much neater. I have fixed it, PTAL

m1a2st

Thank @xijiu for this patch, a little comment

m1a2st · 2025-07-15T14:02:53Z

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)


Nit: No need to add @Timeout here, @ClusterTest already includes a timeout setting.

@m1a2st Thanks very much, the @Timeout is redundant, and I have remove it, PTAL

TaiJuWu

LGTM.

AndrewJSchofield

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.

xijiu · 2025-07-17T07:03:13Z

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

AndrewJSchofield

Thanks for the updates. Just a few minor comments now.

AndrewJSchofield · 2025-07-18T08:52:32Z

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

+            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");


nit: Grammar "do not exist" I think.

AndrewJSchofield · 2025-07-18T08:52:40Z

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

                throw new LeaderNotAvailableException("The partitions \"" + partitionStr + "\" have no leader");
            }
        }

+        private List<TopicPartition> filterNotExistPartitions(Collection<TopicPartition> topicPartitions) {


nit: filterNonExistentPartitions is probably a bit better.

AndrewJSchofield · 2025-07-18T08:55:10Z

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

+                        .toList();
+
+                return topicPartitions.stream().filter(element -> !existPartitions.contains(element)).toList();
+            } catch (Exception e) {


For consistency, we tend to catch (InterruptedException | ExecutionException e) in this source file.

AndrewJSchofield · 2025-07-18T08:56:17Z

Also please merge latest changes from trunk. Hopefully you'll get a clean test run then.

xijiu · 2025-07-18T17:17:15Z

@AndrewJSchofield Thanks for the code review, and I have fixed them and merged the latest changes from trunk, PTAL

DL1231

LGTM.

AndrewJSchofield

Thanks for the PR. Looks good to me.

chia7712 · 2025-07-22T09:03:51Z

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

+            }
+
+            // check the partitions have leader
+            List<TopicPartition> partitionsWithoutLeader = filterNoneLeaderPartitions(partitionsToReset);


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?

finish coding

157d929

github-actions bot added the triage PRs from the community label Jul 15, 2025

github-actions bot added tools small Small PRs labels Jul 15, 2025

xijiu added ci-approved tools small Small PRs and removed tools small Small PRs labels Jul 15, 2025

fix failed test

5084928

AndrewJSchofield requested changes Jul 15, 2025

View reviewed changes

m1a2st reviewed Jul 15, 2025

View reviewed changes

optimize

edeaa8a

github-actions bot removed the triage PRs from the community label Jul 16, 2025

TaiJuWu approved these changes Jul 16, 2025

View reviewed changes

m1a2st approved these changes Jul 16, 2025

View reviewed changes

AndrewJSchofield requested changes Jul 16, 2025

View reviewed changes

fix

3efa3bb

AndrewJSchofield requested changes Jul 18, 2025

View reviewed changes

xijiu added 2 commits July 19, 2025 00:26

Merge branch 'trunk' of github.com:xijiu/kafka into 19500

bb09668

fix

dc58674

DL1231 approved these changes Jul 20, 2025

View reviewed changes

AndrewJSchofield approved these changes Jul 21, 2025

View reviewed changes

AndrewJSchofield merged commit f188a31 into apache:trunk Jul 21, 2025
24 checks passed

chia7712 reviewed Jul 22, 2025

View reviewed changes

KAFKA-19500: kafka-consumer-groups.sh should fail quickly if the partition leader is unavailable #20168

KAFKA-19500: kafka-consumer-groups.sh should fail quickly if the partition leader is unavailable #20168

Conversation

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

Uh oh!

Uh oh!

xijiu commented Jul 15, 2025

Uh oh!

AndrewJSchofield 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!

m1a2st 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!

TaiJuWu left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

xijiu commented Jul 17, 2025

Uh oh!

AndrewJSchofield 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!

AndrewJSchofield commented Jul 18, 2025

Uh oh!

xijiu commented Jul 18, 2025

Uh oh!

DL1231 left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable #20168

KAFKA-19500: `kafka-consumer-groups.sh` should fail quickly if the partition leader is unavailable #20168

xijiu commented Jul 15, 2025 •

edited by github-actions bot

Loading