Repository: apache/kafka
File: tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java (Java)

Code before:
        } catch (Throwable e) {
            System.err.println("ERROR: " + e);
            e.printStackTrace(System.err);
            return EXIT_CODE_ERROR;
        }
    }

    // visible for testing
    void maybeDeleteActiveConsumers(final String groupId,
                                    final Admin adminClient,
                                    final boolean force)
        throws ExecutionException, InterruptedException {
        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) {
                        continue;
                    }
                }
                throw ee;
            }
        }
    }

    private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map<Object, Object> consumerConfig,
                                                                    final StreamsResetterOptions options)
        throws IOException, ParseException {

Review comment (performance):
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 🤔 

Code after:
                return exitCode;
            }
        } catch (Throwable e) {
            System.err.println("ERROR: " + e);
            e.printStackTrace(System.err);
            return EXIT_CODE_ERROR;
        }
    }

    // visible for testing
    void maybeDeleteActiveConsumers(final String groupId,
                                    final Admin adminClient,
                                    final boolean force)
        throws ExecutionException, InterruptedException {
        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.
                if (ee.getCause() instanceof KafkaException ke && ke.getCause() instanceof UnknownMemberIdException) {
                    if (retries++ < MAX_REMOVE_MEMBERS_FROM_CONSUMER_GROUP_RETRIES) {
                        continue;
                    }
                }
                throw ee;
            }
        }
    }

    private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map<Object, Object> consumerConfig,