You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org> on 2025/12/04 00:37:18 UTC
[PR] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Nikita-Shupletsov opened a new pull request, #21074:
URL: https://github.com/apache/kafka/pull/21074
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.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2591530172
##########
tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java:
##########
@@ -293,6 +313,102 @@ public void testResetToDatetimeWhenPartitionIsEmptyResetsToLatestOffset() {
assertEquals(beginningAndEndOffset, position);
}
+ @Test
+ public void shouldRetryToRemoveMembersOnUnknownMemberIdExceptionAndForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of(new LeaveGroupRequestData.MemberIdentity(), Errors.UNKNOWN_MEMBER_ID)), Set.of()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of()), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient, times(2)).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldFailAfterTooManyRetries() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of(new LeaveGroupRequestData.MemberIdentity(), Errors.UNKNOWN_MEMBER_ID)), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ assertThrows(ExecutionException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true));
+
+ verify(adminClient, times(4)).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldFailIfThereAreMembersAndNotForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ assertThrows(IllegalStateException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, false));
+
+ verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldRemoveIfThereAreMembersAndForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of()), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldIgnoreGroupIdNotFoundException() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ final KafkaFutureImpl<ConsumerGroupDescription> future = new KafkaFutureImpl<>();
+ future.completeExceptionally(new GroupIdNotFoundException(groupId));
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, future)));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ 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);
Review Comment:
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
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "gensericghiro (via GitHub)" <gi...@apache.org>.
gensericghiro commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2589955505
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +181,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
+ + "and has following members: " + members + ". "
+ + "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.");
+ }
+ }
+ break;
+ } catch (ExecutionException ee) {
+ // If the group ID is not found, this is not an error case
+ if (ee.getCause() instanceof GroupIdNotFoundException) {
+ break;
+ }
+ // if a member is unknown, it may mean that it left the group itself. Retrying to confirm.
Review Comment:
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?
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #21074:
URL: https://github.com/apache/kafka/pull/21074
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587124614
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
Review Comment:
why not use `while(retries++ < 3)` ?
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "gensericghiro (via GitHub)" <gi...@apache.org>.
gensericghiro commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2603198359
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +181,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
+ + "and has following members: " + members + ". "
+ + "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.");
+ }
+ }
+ break;
+ } catch (ExecutionException ee) {
+ // If the group ID is not found, this is not an error case
+ if (ee.getCause() instanceof GroupIdNotFoundException) {
+ break;
+ }
+ // if a member is unknown, it may mean that it left the group itself. Retrying to confirm.
Review Comment:
I see, didn't know `removeMembersFromConsumerGroup` fetched and deleted members.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587167108
##########
tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java:
##########
@@ -293,6 +310,77 @@ public void testResetToDatetimeWhenPartitionIsEmptyResetsToLatestOffset() {
assertEquals(beginningAndEndOffset, position);
}
+ @Test
+ public void shouldRetryToRemoveMembersOnUnknownMemberIdExceptionAndForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(new RemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of(new LeaveGroupRequestData.MemberIdentity(), Errors.UNKNOWN_MEMBER_ID)), Set.of()))
+ .thenReturn(new RemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of()), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient, times(2)).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldFailIfThereAreMembersAndNotForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ assertThrows(IllegalStateException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, false));
+
+ verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());
Review Comment:
This is because the `force` parameter is set to `false`, right?
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
Review Comment:
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 🤔
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
+ + "and has following members: " + members + ". "
+ + "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.");
+ }
+ }
+ return;
+ } catch (ExecutionException ee) {
+ // If the group ID is not found, this is not an error case
+ if (ee.getCause() instanceof GroupIdNotFoundException) {
+ return;
+ }
+ // 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) {
Review Comment:
Consider adding some kind of constant like `NUM_RETRIES` or something.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2591057112
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
Review Comment:
Ah. The diff was hard to read. I see now.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587874050
##########
tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java:
##########
@@ -293,6 +310,77 @@ public void testResetToDatetimeWhenPartitionIsEmptyResetsToLatestOffset() {
assertEquals(beginningAndEndOffset, position);
}
+ @Test
+ public void shouldRetryToRemoveMembersOnUnknownMemberIdExceptionAndForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(new RemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of(new LeaveGroupRequestData.MemberIdentity(), Errors.UNKNOWN_MEMBER_ID)), Set.of()))
+ .thenReturn(new RemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of()), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient, times(2)).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldFailIfThereAreMembersAndNotForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ assertThrows(IllegalStateException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, false));
+
+ verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());
Review Comment:
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.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2589273481
##########
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java:
##########
@@ -35,7 +35,7 @@ public class RemoveMembersFromConsumerGroupResult {
private final KafkaFuture<Map<MemberIdentity, Errors>> future;
private final Set<MemberToRemove> memberInfos;
- RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
+ public RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
Set<MemberToRemove> memberInfos) {
Review Comment:
I would say we cannot. This is public API https://kafka.apache.org/41/javadoc/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.html
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21074:
URL: https://github.com/apache/kafka/pull/21074#issuecomment-3639543483
Thanks for the PR. Merged to `trunk`.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587130213
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
+ + "and has following members: " + members + ". "
+ + "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.");
+ }
+ }
+ return;
+ } catch (ExecutionException ee) {
+ // If the group ID is not found, this is not an error case
+ if (ee.getCause() instanceof GroupIdNotFoundException) {
+ return;
Review Comment:
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?
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2590048590
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +181,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
+ + "and has following members: " + members + ". "
+ + "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.");
+ }
+ }
+ break;
+ } catch (ExecutionException ee) {
+ // If the group ID is not found, this is not an error case
+ if (ee.getCause() instanceof GroupIdNotFoundException) {
+ break;
+ }
+ // if a member is unknown, it may mean that it left the group itself. Retrying to confirm.
Review Comment:
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
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +181,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
+ + "and has following members: " + members + ". "
+ + "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.");
+ }
+ }
+ break;
+ } catch (ExecutionException ee) {
+ // If the group ID is not found, this is not an error case
+ if (ee.getCause() instanceof GroupIdNotFoundException) {
+ break;
+ }
+ // if a member is unknown, it may mean that it left the group itself. Retrying to confirm.
Review Comment:
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
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587861240
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
Review Comment:
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
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2589771575
##########
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java:
##########
@@ -35,7 +35,7 @@ public class RemoveMembersFromConsumerGroupResult {
private final KafkaFuture<Map<MemberIdentity, Errors>> future;
private final Set<MemberToRemove> memberInfos;
- RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
+ public RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
Set<MemberToRemove> memberInfos) {
Review Comment:
thanks. as it's done only for the test, I will modify 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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2591066486
##########
tools/src/test/java/org/apache/kafka/tools/StreamsResetterTest.java:
##########
@@ -293,6 +313,102 @@ public void testResetToDatetimeWhenPartitionIsEmptyResetsToLatestOffset() {
assertEquals(beginningAndEndOffset, position);
}
+ @Test
+ public void shouldRetryToRemoveMembersOnUnknownMemberIdExceptionAndForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of(new LeaveGroupRequestData.MemberIdentity(), Errors.UNKNOWN_MEMBER_ID)), Set.of()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of()), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient, times(2)).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldFailAfterTooManyRetries() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of(new LeaveGroupRequestData.MemberIdentity(), Errors.UNKNOWN_MEMBER_ID)), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ assertThrows(ExecutionException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true));
+
+ verify(adminClient, times(4)).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldFailIfThereAreMembersAndNotForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ assertThrows(IllegalStateException.class, () -> streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, false));
+
+ verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldRemoveIfThereAreMembersAndForce() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, KafkaFutureImpl.completedFuture(consumerGroupDescription))));
+ when(adminClient.removeMembersFromConsumerGroup(eq(groupId), any()))
+ .thenReturn(createRemoveMembersFromConsumerGroupResult(KafkaFutureImpl.completedFuture(Map.of()), Set.of()));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ @Test
+ public void shouldIgnoreGroupIdNotFoundException() throws Exception {
+ final String groupId = "groupId";
+
+ final Admin adminClient = mock(Admin.class);
+ final ConsumerGroupDescription consumerGroupDescription = mock(ConsumerGroupDescription.class);
+
+ final KafkaFutureImpl<ConsumerGroupDescription> future = new KafkaFutureImpl<>();
+ future.completeExceptionally(new GroupIdNotFoundException(groupId));
+ when(adminClient.describeConsumerGroups(eq(Set.of(groupId)), any()))
+ .thenReturn(new DescribeConsumerGroupsResult(Map.of(groupId, future)));
+ when(consumerGroupDescription.members()).thenReturn(List.of(mock(MemberDescription.class)));
+
+ streamsResetter.maybeDeleteActiveConsumers(groupId, adminClient, true);
+
+ verify(adminClient, never()).removeMembersFromConsumerGroup(eq(groupId), any());
+ }
+
+ 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);
Review Comment:
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?
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587872082
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
Review Comment:
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
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587124614
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
Review Comment:
why not use `while(retries < 3)` ?
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587120711
##########
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java:
##########
@@ -35,7 +35,7 @@ public class RemoveMembersFromConsumerGroupResult {
private final KafkaFuture<Map<MemberIdentity, Errors>> future;
private final Set<MemberToRemove> memberInfos;
- RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
+ public RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
Set<MemberToRemove> memberInfos) {
Review Comment:
@lianetm -- can we make this constructor public w/o a KIP?
nit: fix indention for second parmeter.
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2587861240
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
Review Comment:
because I wanted to still through 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
--
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] MINOR: added retries if removeMembersFromConsumerGroup failed with UnknownMemberIdException [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #21074:
URL: https://github.com/apache/kafka/pull/21074#discussion_r2590944569
##########
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java:
##########
@@ -177,30 +179,42 @@ public int execute(final String[] args, final Properties config) {
}
}
- private void maybeDeleteActiveConsumers(final String groupId,
- final Admin adminClient,
- final StreamsResetterOptions options)
+ // visible for testing
+ void maybeDeleteActiveConsumers(final String groupId,
+ final Admin adminClient,
+ final boolean force)
throws ExecutionException, InterruptedException {
- final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
- Set.of(groupId),
- new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
- try {
- final List<MemberDescription> members =
- new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
- if (!members.isEmpty()) {
- if (options.hasForce()) {
- System.out.println("Force deleting all active members in the group: " + groupId);
- adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
- } else {
- throw new IllegalStateException("Consumer group '" + groupId + "' is still active "
- + "and has following members: " + members + ". "
- + "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) {
+ final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
+ Set.of(groupId),
+ new DescribeConsumerGroupsOptions().timeoutMs(10 * 1000));
+ try {
+ final List<MemberDescription> members =
+ new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
+ if (!members.isEmpty()) {
+ if (force) {
+ System.out.println("Force deleting all active members in the group: " + groupId);
+ adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions()).all().get();
Review Comment:
That makes sense.
It seems a bit odd because while `MemberToRemove` doesn't support member IDs, all `removeMembersFromConsumerGroup()` does with the `MemberRemove`s is to convert each of them into a `MemberIdentity` which _does_ support member IDs 🤷♂️
Thanks for the explanation 👍
--
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