KAFKA-9146: KIP-571 Add option to force delete active members in StreamsResetter

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -182,7 +182,6 @@
 import org.apache.kafka.common.requests.FindCoordinatorResponse;
 import org.apache.kafka.common.requests.IncrementalAlterConfigsRequest;
 import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
-import org.apache.kafka.common.requests.JoinGroupRequest;
 import org.apache.kafka.common.requests.LeaveGroupRequest;
 import org.apache.kafka.common.requests.LeaveGroupResponse;
 import org.apache.kafka.common.requests.ListGroupsRequest;
@@ -3612,6 +3611,27 @@ private boolean dependsOnSpecificNode(ConfigResource resource) {
                 || resource.type() == ConfigResource.Type.BROKER_LOGGER;
     }
 
+    private List<MemberIdentity> getMembersFromGroup(String groupId) {
+        Collection<MemberDescription> members = new ArrayList<>();
+        try {
+            members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
+        } catch (Throwable ex) {
+            System.out.println("Encounter exception when trying to get members from group: " + groupId);
+            ex.printStackTrace();
+        }
+
+        List<MemberIdentity> memberToRemove = new ArrayList<>();
+        for (MemberDescription member: members) {
+            if (member.groupInstanceId().isPresent()) {
+                memberToRemove.add(new MemberIdentity().setGroupInstanceId(member.groupInstanceId().get())
+                );
+            } else {
+                memberToRemove.add(new MemberIdentity().setMemberId(member.consumerId()));
+            }
+        }
+        return memberToRemove;
+    }
+
     @Override
     public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(String groupId,
                                                                                RemoveMembersFromConsumerGroupOptions options) {
@@ -3621,24 +3641,37 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
         KafkaFutureImpl<Map<MemberIdentity, Errors>> future = new KafkaFutureImpl<>();
 
         ConsumerGroupOperationContext<Map<MemberIdentity, Errors>, RemoveMembersFromConsumerGroupOptions> context =
-            new ConsumerGroupOperationContext<>(groupId, options, deadline, future);
+                new ConsumerGroupOperationContext<>(groupId, options, deadline, future);
 
-        Call findCoordinatorCall = getFindCoordinatorCall(context,
-            () -> getRemoveMembersFromGroupCall(context));
+        Call findCoordinatorCall;
+        if (options.removeAll()) {
+            List<MemberIdentity> members = getMembersFromGroup(groupId);
+            findCoordinatorCall = getFindCoordinatorCall(context,
+                () -> getRemoveMembersFromGroupCall(context, members));
+        } else {
+            findCoordinatorCall = getFindCoordinatorCall(context,
+                () -> getRemoveMembersFromGroupCall(context, new ArrayList<>()));
+        }
         runnable.call(findCoordinatorCall, startFindCoordinatorMs);
 
-        return new RemoveMembersFromConsumerGroupResult(future, options.members());
+        return new RemoveMembersFromConsumerGroupResult(future, options.members(), options.removeAll());
     }
 
-    private Call getRemoveMembersFromGroupCall(ConsumerGroupOperationContext<Map<MemberIdentity, Errors>, RemoveMembersFromConsumerGroupOptions> context) {
+    private Call getRemoveMembersFromGroupCall(ConsumerGroupOperationContext<Map<MemberIdentity, Errors>,
+            RemoveMembersFromConsumerGroupOptions> context, List<MemberIdentity> allMembers) {
         return new Call("leaveGroup",
                         context.deadline(),
                         new ConstantNodeIdProvider(context.node().get().id())) {
             @Override
             LeaveGroupRequest.Builder createRequest(int timeoutMs) {
-                return new LeaveGroupRequest.Builder(context.groupId(),
-                                                     context.options().members().stream().map(
-                                                         MemberToRemove::toMemberIdentity).collect(Collectors.toList()));
+                if (context.options().removeAll()) {
+                    return new LeaveGroupRequest.Builder(context.groupId(),
+                            allMembers);
+                } else {
+                    return new LeaveGroupRequest.Builder(context.groupId(),
+                            context.options().members().stream().map(
+                                    MemberToRemove::toMemberIdentity).collect(Collectors.toList()));
+                }
             }
 
             @Override
@@ -3647,7 +3680,7 @@ void handleResponse(AbstractResponse abstractResponse) {
 
                 // If coordinator changed since we fetched it, retry
                 if (ConsumerGroupOperationContext.hasCoordinatorMoved(response)) {
-                    Call call = getRemoveMembersFromGroupCall(context);
+                    Call call = getRemoveMembersFromGroupCall(context, allMembers);
                     rescheduleFindCoordinatorTask(context, () -> call, this);
                     return;
                 }
@@ -3660,7 +3693,7 @@ void handleResponse(AbstractResponse abstractResponse) {
                     // We set member.id to empty here explicitly, so that the lookup will succeed as user doesn't
                     // know the exact member.id.
                     memberErrors.put(new MemberIdentity()
-                                         .setMemberId(JoinGroupRequest.UNKNOWN_MEMBER_ID)
+                                         .setMemberId(memberResponse.memberId())
                                          .setGroupInstanceId(memberResponse.groupInstanceId()),
                                      Errors.forCode(memberResponse.errorCode()));
                 }

clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupOptions.java

@@ -32,12 +32,23 @@
 public class RemoveMembersFromConsumerGroupOptions extends AbstractOptions<RemoveMembersFromConsumerGroupOptions> {
 
     private Set<MemberToRemove> members;
+    private boolean removeAll;
 
     public RemoveMembersFromConsumerGroupOptions(Collection<MemberToRemove> members) {
         this.members = new HashSet<>(members);
+        this.removeAll = false;
+    }
+
+    public RemoveMembersFromConsumerGroupOptions(Boolean removeAll) {
+        this.members = new HashSet<>();
+        this.removeAll = removeAll;
     }
 
     public Set<MemberToRemove> members() {
         return members;
     }
+
+    public boolean removeAll() {
+        return removeAll;
+    }
 }

clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java

@@ -33,11 +33,14 @@ public class RemoveMembersFromConsumerGroupResult {
 
     private final KafkaFuture<Map<MemberIdentity, Errors>> future;
     private final Set<MemberToRemove> memberInfos;
+    private final boolean removeAll;
 
     RemoveMembersFromConsumerGroupResult(KafkaFuture<Map<MemberIdentity, Errors>> future,
-                                         Set<MemberToRemove> memberInfos) {
+                                         Set<MemberToRemove> memberInfos,
+                                         Boolean removeAll) {
         this.future = future;
         this.memberInfos = memberInfos;
+        this.removeAll = removeAll;
     }
 
     /**
@@ -46,20 +49,33 @@ public class RemoveMembersFromConsumerGroupResult {
      * If not, the first member error shall be returned.
      */
     public KafkaFuture<Void> all() {
-        final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
-        this.future.whenComplete((memberErrors, throwable) -> {
-            if (throwable != null) {
-                result.completeExceptionally(throwable);
-            } else {
-                for (MemberToRemove memberToRemove : memberInfos) {
-                    if (maybeCompleteExceptionally(memberErrors, memberToRemove.toMemberIdentity(), result)) {
-                        return;
+        if (removeAll) {
+            final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+            this.future.whenComplete((memberErrors, throwable) -> {
+                if (throwable != null) {
+                    result.completeExceptionally(throwable);
+                } else {
+                    System.out.println("Remove all active members succeeded, removed " + memberErrors.size() + " members: " + memberErrors.keySet());
+                    result.complete(null);
+                }
+            });
+            return result;
+        } else {
+            final KafkaFutureImpl<Void> result = new KafkaFutureImpl<>();
+            this.future.whenComplete((memberErrors, throwable) -> {
+                if (throwable != null) {
+                    result.completeExceptionally(throwable);
+                } else {
+                    for (MemberToRemove memberToRemove : memberInfos) {
+                        if (maybeCompleteExceptionally(memberErrors, memberToRemove.toMemberIdentity(), result)) {
+                            return;
+                        }
                     }
+                    result.complete(null);
                 }
-                result.complete(null);
-            }
-        });
-        return result;
+            });
+            return result;
+        }
     }
 
     /**

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -2411,6 +2411,50 @@ public void testRemoveMembersFromGroup() throws Exception {
             assertNull(noErrorResult.all().get());
             assertNull(noErrorResult.memberResult(memberOne).get());
             assertNull(noErrorResult.memberResult(memberTwo).get());
+
+            // Return with success for "removeAll" scenario
+            // 1. KafkaAdminClient should query all members successfully
+            // 1.1 construct the DescribeGroupsResponse
+            TopicPartition myTopicPartition0 = new TopicPartition("my_topic", 0);
+            TopicPartition myTopicPartition1 = new TopicPartition("my_topic", 1);
+            TopicPartition myTopicPartition2 = new TopicPartition("my_topic", 2);
+
+            final List<TopicPartition> topicPartitions = new ArrayList<>();
+            topicPartitions.add(0, myTopicPartition0);
+            topicPartitions.add(1, myTopicPartition1);
+            topicPartitions.add(2, myTopicPartition2);
+
+            final ByteBuffer memberAssignment = ConsumerProtocol.serializeAssignment(new ConsumerPartitionAssignor.Assignment(topicPartitions));
+            byte[] memberAssignmentBytes = new byte[memberAssignment.remaining()];
+            DescribedGroupMember groupInstanceOne = DescribeGroupsResponse.groupMember("0", instanceOne, "clientId0", "clientHost", memberAssignmentBytes, null);
+            DescribedGroupMember groupInstanceTwo = DescribeGroupsResponse.groupMember("1", instanceTwo, "clientId1", "clientHost", memberAssignmentBytes, null);
+            DescribeGroupsResponseData data = new DescribeGroupsResponseData();
+            data.groups().add(DescribeGroupsResponse.groupMetadata(
+                    groupId,
+                    Errors.NONE,
+                    "",
+                    ConsumerProtocol.PROTOCOL_TYPE,
+                    "",
+                    asList(groupInstanceOne, groupInstanceTwo),
+                    Collections.emptySet()));
+
+            // 1.2 prepare response for AdminClient.describeConsumerGroups
+            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));
+            env.kafkaClient().prepareResponse(new DescribeGroupsResponse(data));
+
+            // 2. KafkaAdminClient should delete all members correctly
+            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));
+            env.kafkaClient().prepareResponse(new LeaveGroupResponse(
+                    new LeaveGroupResponseData().setErrorCode(Errors.NONE.code()).setMembers(
+                            Arrays.asList(responseTwo,
+                                    new MemberResponse().setGroupInstanceId(instanceOne).setErrorCode(Errors.NONE.code())
+                            ))
+            ));
+            final RemoveMembersFromConsumerGroupResult successResult = env.adminClient().removeMembersFromConsumerGroup(
+                    groupId,
+                    new RemoveMembersFromConsumerGroupOptions(true)
+            );
+            assertNull(successResult.all().get());
         }
     }
 

clients/src/test/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResultTest.java

@@ -61,7 +61,7 @@ public void setUp() {
     public void testTopLevelErrorConstructor() throws InterruptedException {
         memberFutures.completeExceptionally(Errors.GROUP_AUTHORIZATION_FAILED.exception());
         RemoveMembersFromConsumerGroupResult topLevelErrorResult =
-            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
+            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove, false);
         TestUtils.assertFutureError(topLevelErrorResult.all(), GroupAuthorizationException.class);
     }
 
@@ -76,7 +76,7 @@ public void testMemberMissingErrorInRequestConstructor() throws InterruptedExcep
         memberFutures.complete(errorsMap);
         assertFalse(memberFutures.isCompletedExceptionally());
         RemoveMembersFromConsumerGroupResult missingMemberResult =
-            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
+            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove, false);
 
         TestUtils.assertFutureError(missingMemberResult.all(), IllegalArgumentException.class);
         assertNull(missingMemberResult.memberResult(instanceOne).get());
@@ -97,7 +97,7 @@ public void testNoErrorConstructor() throws ExecutionException, InterruptedExcep
         errorsMap.put(instanceOne.toMemberIdentity(), Errors.NONE);
         errorsMap.put(instanceTwo.toMemberIdentity(), Errors.NONE);
         RemoveMembersFromConsumerGroupResult noErrorResult =
-            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
+            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove, false);
         memberFutures.complete(errorsMap);
 
         assertNull(noErrorResult.all().get());
@@ -109,7 +109,7 @@ private RemoveMembersFromConsumerGroupResult createAndVerifyMemberLevelError() t
         memberFutures.complete(errorsMap);
         assertFalse(memberFutures.isCompletedExceptionally());
         RemoveMembersFromConsumerGroupResult memberLevelErrorResult =
-            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove);
+            new RemoveMembersFromConsumerGroupResult(memberFutures, membersToRemove, false);
 
         TestUtils.assertFutureError(memberLevelErrorResult.all(), FencedInstanceIdException.class);
         assertNull(memberLevelErrorResult.memberResult(instanceOne).get());

core/src/main/scala/kafka/tools/StreamsResetter.java

@@ -27,6 +27,7 @@
 import org.apache.kafka.clients.admin.DescribeConsumerGroupsOptions;
 import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
 import org.apache.kafka.clients.admin.MemberDescription;
+import org.apache.kafka.clients.admin.RemoveMembersFromConsumerGroupOptions;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -106,6 +107,7 @@ public class StreamsResetter {
     private static OptionSpec versionOption;
     private static OptionSpecBuilder executeOption;
     private static OptionSpec<String> commandConfigOption;
+    private static OptionSpec forceOption;
 
     private static String usage = "This tool helps to quickly reset an application in order to reprocess "
             + "its data from scratch.\n"
@@ -149,7 +151,7 @@ public int run(final String[] args,
             properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServerOption));
 
             adminClient = Admin.create(properties);
-            validateNoActiveConsumers(groupId, adminClient);
+            maybeDeleteActiveConsumers(groupId, adminClient);
 
             allTopics.clear();
             allTopics.addAll(adminClient.listTopics().names().get(60, TimeUnit.SECONDS));
@@ -176,8 +178,8 @@ public int run(final String[] args,
         return exitCode;
     }
 
-    private void validateNoActiveConsumers(final String groupId,
-                                           final Admin adminClient)
+    private void maybeDeleteActiveConsumers(final String groupId,
+                                            final Admin adminClient)
         throws ExecutionException, InterruptedException {
 
         final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(
@@ -186,9 +188,15 @@ private void validateNoActiveConsumers(final String groupId,
         final List<MemberDescription> members =
             new ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
         if (!members.isEmpty()) {
-            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.");
+            if (options.has(forceOption)) {
+                System.out.println("Force deleting all active members in the group: " + groupId);
+                adminClient.removeMembersFromConsumerGroup(groupId, new RemoveMembersFromConsumerGroupOptions(true)).all();
+            } 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." +
+                        "Try set '--force' in the cmdline to force delete active members.");
+            }
         }
     }
 
@@ -236,6 +244,8 @@ private void parseArguments(final String[] args) {
             .withRequiredArg()
             .ofType(String.class)
             .describedAs("file name");
+        forceOption = optionParser.accepts("force", "Force remove members when long session time out has been configured, " +
+                "please make sure to shut down all stream applications when this option is specified to avoid unexpected rebalances.");
         executeOption = optionParser.accepts("execute", "Execute the command.");
         dryRunOption = optionParser.accepts("dry-run", "Display the actions that would be performed without executing the reset commands.");
         helpOption = optionParser.accepts("help", "Print usage information.").forHelp();