[GitHub] [kafka] abbccdda commented on a change in pull request #8589: KAFKA-9146: KIP-571 Add option to force delete active members in StreamsResetter

[GitBox]

abbccdda commented on a change in pull request #8589:
URL: https://github.com/apache/kafka/pull/8589#discussion_r418108095



##########
File path: 
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -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));

Review comment:
       could we pass the members into the context?

##########
File path: 
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;

Review comment:
       Could we just make members to be `Optional<Set<MemberToRemove>>` so that 
we don't need a separate removeAll parameter?

##########
File path: 
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -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) {

Review comment:
       style error here.
   
   I would recommend doing a self style check like:
   `./gradlew checkstyleMain checkstyleTest spotbugsMain spotbugsTest 
spotbugsScoverage compileTestJava` otherwise we still need to fix those 
failures after we do jenkins build.

##########
File path: 
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -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);

Review comment:
       Remove print statements

##########
File path: 
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -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();

Review comment:
       Curious why we are still continuing in this case, as the member lookup 
already fails.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org