KIP-222 - Add Consumer Group operations to Admin API

checkstyle/checkstyle.xml

@@ -114,7 +114,7 @@
 
     <module name="ClassFanOutComplexity">
       <!-- default is 20 -->
-      <property name="max" value="40"/>
+      <property name="max" value="50"/>
     </module>
     <module name="CyclomaticComplexity">
       <!-- default is 10-->

checkstyle/import-control.xml

@@ -163,6 +163,8 @@
 
     <subpackage name="admin">
       <allow pkg="org.apache.kafka.clients.admin" />
+      <allow pkg="org.apache.kafka.clients.consumer.internals" />
+      <allow pkg="org.apache.kafka.clients.consumer" />
     </subpackage>
   </subpackage>
 

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

@@ -535,4 +535,86 @@ public DeleteRecordsResult deleteRecords(Map<TopicPartition, RecordsToDelete> re
      */
     public abstract DeleteRecordsResult deleteRecords(Map<TopicPartition, RecordsToDelete> recordsToDelete,
                                                       DeleteRecordsOptions options);
+    /**
+     * Describe some group IDs in the cluster.
+     *
+     * @param groupIds The IDs of the groups to describe.
+     * @param options  The options to use when describing the groups.
+     * @return The DescribeConsumerGroupResult.
+     */
+    public abstract DescribeGroupsResult describeGroups(Collection<String> groupIds,
+                                                        DescribeGroupsOptions options);
+
+    /**
+     * Describe some group IDs in the cluster, with the default options.
+     * <p>
+     * This is a convenience method for
+     * #{@link AdminClient#describeGroups(Collection, DescribeGroupsOptions)} with
+     * default options. See the overload for more details.
+     *
+     * @param groupIds The IDs of the groups to describe.
+     * @return The DescribeConsumerGroupResult.
+     */
+    public DescribeGroupsResult describeGroups(Collection<String> groupIds) {
+        return describeGroups(groupIds, new DescribeGroupsOptions());
+    }
+
+    /**
+     * List the groups available in the cluster.
+     *
+     * @param options           The options to use when listing the groups.
+     * @return The ListGroupsResult.
+     */
+    public abstract ListGroupsResult listGroups(ListGroupsOptions options);
+
+    /**
+     * List the groups available in the cluster with the default options.
+     *
+     * This is a convenience method for #{@link AdminClient#listGroups(ListGroupsOptions)} with default options.
+     * See the overload for more details.
+     *
+     * @return The ListGroupsResult.
+     */
+    public ListGroupsResult listGroups() {
+        return listGroups(new ListGroupsOptions());
+    }
+
+    /**
+     * List the consumer groups available in the cluster.
+     *
+     * @param options           The options to use when listing the groups.
+     * @return The ListGroupsResult.
+     */
+    public abstract ListGroupsResult listConsumerGroups(ListGroupsOptions options);
+
+    /**
+     * List the consumer groups available in the cluster with the default options.
+     *
+     * This is a convenience method for #{@link AdminClient#listConsumerGroups(ListGroupsOptions)} with default options.
+     * See the overload for more details.
+     *
+     * @return The ListGroupsResult.
+     */
+    public ListGroupsResult listConsumerGroups() {
+        return listConsumerGroups(new ListGroupsOptions());
+    }
+
+    /**
+     * List the group offsets available in the cluster.
+     *
+     * @param options           The options to use when listing the group offsets.
+     * @return The ListGroupOffsetsResult
+     */
+    public abstract ListGroupOffsetsResult listGroupOffsets(String groupId, ListGroupOffsetsOptions options);
+
+    /**
+     * List the group offsets available in the cluster with the default options.
+     *
+     * This is a convenience method for #{@link AdminClient#listGroupOffsets(String, ListGroupOffsetsOptions)} with default options.
+     *
+     * @return The ListGroupOffsetsResult.
+     */
+    public ListGroupOffsetsResult listGroupOffsets(String groupId) {
+        return listGroupOffsets(groupId, new ListGroupOffsetsOptions());
+    }
 }

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

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+
+/**
+ * Options for {@link AdminClient#describeGroups(Collection, DescribeGroupsOptions)}.
+ * <p>
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeGroupsOptions extends AbstractOptions<DescribeGroupsOptions> {
+
+}

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

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+
+/**
+ * The result of the {@link KafkaAdminClient#describeGroups(Collection, DescribeGroupsOptions)}} call.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class DescribeGroupsResult {
+
+    private final Map<String, KafkaFuture<GroupDescription>> futures;
+
+    public DescribeGroupsResult(Map<String, KafkaFuture<GroupDescription>> futures) {
+        this.futures = futures;
+    }
+
+    /**
+     * Return a map from group name to futures which can be used to check the description of group.
+     */
+    public Map<String, KafkaFuture<GroupDescription>> values() {
+        return futures;
+    }
+
+    /**
+     * Return a future which succeeds only if all the group descriptions succeed.
+     */
+    public KafkaFuture<Map<String, GroupDescription>> all() {
+        return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).
+            thenApply(new KafkaFuture.Function<Void, Map<String, GroupDescription>>() {
+                @Override
+                public Map<String, GroupDescription> apply(Void v) {
+                    Map<String, GroupDescription> descriptions = new HashMap<>(futures.size());
+                    for (Map.Entry<String, KafkaFuture<GroupDescription>> entry : futures.entrySet()) {
+                        try {
+                            descriptions.put(entry.getKey(), entry.getValue().get());
+                        } catch (InterruptedException | ExecutionException e) {
+                            // This should be unreachable, because allOf ensured that all the futures
+                            // completed successfully.
+                            throw new RuntimeException(e);
+                        }
+                    }
+                    return descriptions;
+                }
+            });
+    }
+
+}

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

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.utils.Utils;
+
+import java.util.List;
+
+/**
+ * A detailed description of a single consumer group in the cluster.
+ */
+public class GroupDescription {
+
+    private final String groupId;
+    private final String protocolType;
+    private final List<MemberDescription> members;
+
+    public GroupDescription(String groupId, String protocolType, List<MemberDescription> members) {
+        this.groupId = groupId;
+        this.protocolType = protocolType;
+        this.members = members;
+    }
+
+    public String groupId() {
+        return groupId;
+    }
+
+    public String protocolType() {
+        return protocolType;
+    }
+
+    public List<MemberDescription> members() {
+        return members;
+    }
+
+    @Override
+    public String toString() {
+        return "(groupId=" + groupId + ", protocolType=" + protocolType + ", members=" +
+            Utils.join(members, ",") + ")";
+    }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+/**
+ * A listing of a group in the cluster.
+ */
+public class GroupListing {
+    private final String protocolType;
+
+    /**
+     * Create an instance with the specified parameters.
+     *
+     * @param protocolType The group protocol type
+     */
+    public GroupListing(String protocolType) {
+        this.protocolType = protocolType;
+    }
+
+    /**
+     * The protocol type of the group.
+     */
+    public String protocolType() {
+        return protocolType;
+    }
+
+    @Override
+    public String toString() {
+        return "(protocolType=" + protocolType + ")";
+    }
+}