MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException #21074

Nikita-Shupletsov · 2025-12-04T00:37:15Z

While we are trying to delete members from a group, they may leave the
group themselves. Which would lead to a failure of the tool, even though
not having that member in the group is the desired state.

…knownMemberIdException.

mjsax · 2025-12-04T01:36:04Z

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


-    RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
+    public RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
                                         Set<MemberToRemove> memberInfos) {


@lianetm -- can we make this constructor public w/o a KIP?

nit: fix indention for second parmeter.

I would say we cannot. This is public API https://kafka.apache.org/41/javadoc/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.html

thanks. as it's done only for the test, I will modify it

mjsax · 2025-12-04T01:37:34Z

tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java

-                        + "Make sure to stop all running application instances before running the reset tool."
-                        + " You can use option '--force' to remove active members from the group.");
+        int retries = 0;
+        while (true) {


why not use while(retries++ < 3) ?

because I wanted to still throw the exception if we are still getting the exception after the retries. we can do while(retries++ < 3), but we would need to either store the exception or throw something generic, which would hide the original exception. so I decided to go this way

Ah. The diff was hard to read. I see now.

mjsax · 2025-12-04T01:39:51Z

tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java

+            } catch (ExecutionException ee) {
+                // If the group ID is not found, this is not an error case
+                if (ee.getCause() instanceof GroupIdNotFoundException) {
+                    return;


Would break be cleaner? Atm, there is no code after the loop, to no difference, but if we add code after the loop in the future, using return here might skip this code accidentally?

kirktrue

Thanks for the PR @Nikita-Shupletsov.

I have mostly questions, as I'm following along to learn more about this area.

Thanks!

kirktrue · 2025-12-04T01:57:26Z

tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java

+                if (!members.isEmpty()) {
+                    if (force) {
+                        System.out.println("Force deleting all active members in the group: " + groupId);
+                        adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();


Unrelated to this PR, but out of curiosity, is there any reason not to pass in the members retrieved in line 193/194 into the RemoveMembersFromConsumerGroupOptions constructor on line 198? That seems like it would save an extra lookup. But for a tool, I guess it's not as critical that it be as efficient as possible 🤔

I was considering it, but if we look at the implementation:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java#L4175-L4178
it passes groupInstanceId if it exists or memberId if it doesn't.
there is no way to replicate this behavior with https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/MemberToRemove.java, as it doesn't accept memberId, only groupInstanceId. and it's a public class, so changing it would require a KIP, afaik

That makes sense.

It seems a bit odd because while MemberToRemove doesn't support member IDs, all removeMembersFromConsumerGroup() does with the MemberRemoves is to convert each of them into a MemberIdentity which does support member IDs 🤷‍♂️

Thanks for the explanation 👍

kirktrue · 2025-12-04T01:59:15Z

tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java

+                }
+                // if a member is unknown, it may mean that it left the group itself. Retrying to confirm.
+                if (ee.getCause() instanceof KafkaException ke && ke.getCause() instanceof UnknownMemberIdException) {
+                    if (retries++ < 3) {


Consider adding some kind of constant like NUM_RETRIES or something.

kirktrue · 2025-12-04T02:01:11Z

tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java

+
+        assertThrows(IllegalStateException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, false));
+
+        verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());


This is because the force parameter is set to false, right?

yes. if it's not force, instead of removing the members, we just fail. as there weren't any tests for that method, I just added this one as well.

gensericghiro · 2025-12-04T17:26:44Z

tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java

+                if (ee.getCause() instanceof GroupIdNotFoundException) {
+                    break;
+                }
+                // if a member is unknown, it may mean that it left the group itself. Retrying to confirm.


I'm probably missing something, but I don't see how retrying 3 times will solve the issue here? Is the hope that on subsequent retries describeConsumerGroups won't return the member that left?

removeMembersFromConsumerGroup fetches all members in the group and sends a request to delete them. if any of them fail, RemoveMembersFromConsumerGroupResult#all will throw an exception.
so, effectively, broker first said something existed, but then said it doesn't.

if we retry, we will repeat the same actions again. and at this point it's not expected to receive the same member(because the broker said it doesn't exit), that failed to get deleted as it was unknown. so even if we skipped some deletion for whatever reason a retry will try to delete it again

…ons in the public API.

mjsax · 2025-12-05T00:59:14Z

tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java

+
+    private RemoveMembersFromConsumerGroupResult createRemoveMembersFromConsumerGroupResult(final KafkaFuture<Map<LeaveGroupRequestData.MemberIdentity, Errors>> future,
+                                                                                            final Set<MemberToRemove> memberInfos) throws Exception {
+        final Constructor<RemoveMembersFromConsumerGroupResult> constructor = RemoveMembersFromConsumerGroupResult.class.getDeclaredConstructor(KafkaFuture.class, Set.class);


Did you explore using MockAdminClient (which atm does not support removeMembersFromConsumerGroup(...) though, but we could maybe add it with this PR), to avoid using reflections?

I did, but it looked like a lot of changes. if you think it's worth it, I would suggest looking at it separately if you don't mind

kirktrue

Looks like there's some linting errors in StreamsResetterTest that are preventing test runs. You can run ./gradlew check -x test locally to perform the same checks.

MINOR: added retries if removeMembersFromConsumerGroup failed with Un…

33d776a

…knownMemberIdException.

github-actions bot added triage PRs from the community tools clients labels Dec 4, 2025

mjsax reviewed Dec 4, 2025

View reviewed changes

kirktrue reviewed Dec 4, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Dec 4, 2025

Addressed comments.

7016288

gensericghiro reviewed Dec 4, 2025

View reviewed changes

Replaced a direct object creation with reflection to avoid modificati…

cac7bb7

…ons in the public API.

lianetm added the ci-approved label Dec 4, 2025

mjsax approved these changes Dec 5, 2025

View reviewed changes

mjsax added streams and removed clients labels Dec 5, 2025

mjsax reviewed Dec 5, 2025

View reviewed changes

kirktrue suggested changes Dec 5, 2025

View reviewed changes

Nikita-Shupletsov added 2 commits December 4, 2025 22:08

Fixed spotlessJavaCheck failure.

681d729

Merge branch 'apache:trunk' into minor-add-retries-to-streams-resetter

38d1106


		assertThrows(IllegalStateException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, false));

		verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());

MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException #21074

Are you sure you want to change the base?

MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException #21074

Conversation

Nikita-Shupletsov commented Dec 4, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Dec 4, 2025 • 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!

Nikita-Shupletsov Dec 4, 2025 • 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!

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

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!

Nikita-Shupletsov Dec 4, 2025 • 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!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Nikita-Shupletsov commented Dec 4, 2025 •

edited by github-actions bot

Loading

mjsax Dec 4, 2025 •

edited

Loading

Nikita-Shupletsov Dec 4, 2025 •

edited

Loading

Nikita-Shupletsov Dec 4, 2025 •

edited

Loading