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