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

Code before:
            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 {
        final List<String> inputTopics = options.inputTopicsOption();
        final List<String> intermediateTopics = options.intermediateTopicsOption();
        int topicNotFound = EXIT_CODE_SUCCESS;

        final List<String> notFoundInputTopics = new ArrayList<>();
        final List<String> notFoundIntermediateTopics = new ArrayList<>();

        if (inputTopics.isEmpty() && intermediateTopics.isEmpty()) {
            System.out.println("No input or intermediate topics specified. Skipping seek.");
            return EXIT_CODE_SUCCESS;
        }

        if (!inputTopics.isEmpty()) {
            System.out.println("Reset-offsets for input topics " + inputTopics);

Review comment (suggestion):
Consider adding some kind of constant like `NUM_RETRIES` or something.

Code after:
        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,
                                                                    final StreamsResetterOptions options)
        throws IOException, ParseException {
        final List<String> inputTopics = options.inputTopicsOption();
        final List<String> intermediateTopics = options.intermediateTopicsOption();
        int topicNotFound = EXIT_CODE_SUCCESS;

        final List<String> notFoundInputTopics = new ArrayList<>();
        final List<String> notFoundIntermediateTopics = new ArrayList<>();

        if (inputTopics.isEmpty() && intermediateTopics.isEmpty()) {
            System.out.println("No input or intermediate topics specified. Skipping seek.");
            return EXIT_CODE_SUCCESS;
        }