KIP-699: Update FindCoordinator to resolve multiple Coordinators at a time

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

@@ -49,13 +49,13 @@ public ConsumerGroupDescription(String groupId,
         this(groupId, isSimpleConsumerGroup, members, partitionAssignor, state, coordinator, Collections.emptySet());
     }
 
-    ConsumerGroupDescription(String groupId,
-                             boolean isSimpleConsumerGroup,
-                             Collection<MemberDescription> members,
-                             String partitionAssignor,
-                             ConsumerGroupState state,
-                             Node coordinator,
-                             Set<AclOperation> authorizedOperations) {
+    public ConsumerGroupDescription(String groupId,
+                                    boolean isSimpleConsumerGroup,
+                                    Collection<MemberDescription> members,
+                                    String partitionAssignor,
+                                    ConsumerGroupState state,
+                                    Node coordinator,
+                                    Set<AclOperation> authorizedOperations) {
         this.groupId = groupId == null ? "" : groupId;
         this.isSimpleConsumerGroup = isSimpleConsumerGroup;
         this.members = members == null ? Collections.emptyList() :

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

@@ -16,10 +16,13 @@
  */
 package org.apache.kafka.clients.admin;
 
+import org.apache.kafka.clients.admin.internals.CoordinatorKey;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -29,9 +32,9 @@
  */
 @InterfaceStability.Evolving
 public class DeleteConsumerGroupsResult {
-    private final Map<String, KafkaFuture<Void>> futures;
+    private final Map<CoordinatorKey, KafkaFutureImpl<Void>> futures;
 
-    DeleteConsumerGroupsResult(final Map<String, KafkaFuture<Void>> futures) {
+    DeleteConsumerGroupsResult(final Map<CoordinatorKey, KafkaFutureImpl<Void>> futures) {
         this.futures = futures;
     }
 
@@ -40,7 +43,9 @@ public class DeleteConsumerGroupsResult {
      * individual deletions.
      */
     public Map<String, KafkaFuture<Void>> deletedGroups() {
-        return futures;
+        Map<String, KafkaFuture<Void>> deletedGroups = new HashMap<>(futures.size());
+        futures.forEach((key, future) -> deletedGroups.put(key.idValue, future));
+        return deletedGroups;
     }
 
     /**

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

@@ -17,10 +17,11 @@
 
 package org.apache.kafka.clients.admin;
 
+import org.apache.kafka.clients.admin.internals.CoordinatorKey;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
 
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -34,39 +35,38 @@
 @InterfaceStability.Evolving
 public class DescribeConsumerGroupsResult {
 
-    private final Map<String, KafkaFuture<ConsumerGroupDescription>> futures;
+    private final Map<CoordinatorKey, KafkaFutureImpl<ConsumerGroupDescription>> futures;
 
-    public DescribeConsumerGroupsResult(final Map<String, KafkaFuture<ConsumerGroupDescription>> futures) {
+    public DescribeConsumerGroupsResult(final Map<CoordinatorKey, KafkaFutureImpl<ConsumerGroupDescription>> futures) {
         this.futures = futures;
     }
 
     /**
      * Return a map from group id to futures which yield group descriptions.
      */
     public Map<String, KafkaFuture<ConsumerGroupDescription>> describedGroups() {
-        return futures;
+        Map<String, KafkaFuture<ConsumerGroupDescription>> describedGroups = new HashMap<>();
+        futures.forEach((key, future) -> describedGroups.put(key.idValue, future));
+        return describedGroups;
     }
 
     /**
      * Return a future which yields all ConsumerGroupDescription objects, if all the describes succeed.
      */
     public KafkaFuture<Map<String, ConsumerGroupDescription>> all() {
         return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
-            new KafkaFuture.BaseFunction<Void, Map<String, ConsumerGroupDescription>>() {
-                @Override
-                public Map<String, ConsumerGroupDescription> apply(Void v) {
+            nil -> {
+                Map<String, ConsumerGroupDescription> descriptions = new HashMap<>(futures.size());
+                futures.forEach((key, future) -> {
                     try {
-                        Map<String, ConsumerGroupDescription> descriptions = new HashMap<>(futures.size());
-                        for (Map.Entry<String, KafkaFuture<ConsumerGroupDescription>> entry : futures.entrySet()) {
-                            descriptions.put(entry.getKey(), entry.getValue().get());
-                        }
-                        return descriptions;
+                        descriptions.put(key.idValue, future.get());
                     } catch (InterruptedException | ExecutionException e) {
                         // This should be unreachable, since the KafkaFuture#allOf already ensured
                         // that all of the futures completed successfully.
                         throw new RuntimeException(e);
                     }
-                }
+                });
+                return descriptions;
             });
     }
 }