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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
