KAFKA-9130: KIP-518 Allow listing consumer groups per state

checkstyle/suppressions.xml

@@ -89,7 +89,7 @@
               files="MockAdminClient.java"/>
 
     <suppress checks="JavaNCSS"
-              files="RequestResponseTest.java|FetcherTest.java"/>
+              files="RequestResponseTest.java|FetcherTest.java|KafkaAdminClientTest.java"/>
 
     <suppress checks="NPathComplexity"
               files="MemoryRecordsTest|MetricsTest"/>

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

@@ -17,12 +17,18 @@
 
 package org.apache.kafka.clients.admin;
 
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.kafka.common.ConsumerGroupState;
+
 /**
  * A listing of a consumer group in the cluster.
  */
 public class ConsumerGroupListing {
     private final String groupId;
     private final boolean isSimpleConsumerGroup;
+    private final Optional<ConsumerGroupState> state;
 
     /**
      * Create an instance with the specified parameters.
@@ -31,8 +37,20 @@ public class ConsumerGroupListing {
      * @param isSimpleConsumerGroup If consumer group is simple or not.
      */
     public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup) {
+        this(groupId, isSimpleConsumerGroup, Optional.empty());
+    }
+
+    /**
+     * Create an instance with the specified parameters.
+     *
+     * @param groupId Group Id
+     * @param isSimpleConsumerGroup If consumer group is simple or not.
+     * @param state The state of the consumer group
+     */
+    public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup, Optional<ConsumerGroupState> state) {
         this.groupId = groupId;
         this.isSimpleConsumerGroup = isSimpleConsumerGroup;
+        this.state = Objects.requireNonNull(state);
     }
 
     /**
@@ -49,11 +67,49 @@ public boolean isSimpleConsumerGroup() {
         return isSimpleConsumerGroup;
     }
 
+    /**
+     * Consumer Group state
+     */
+    public Optional<ConsumerGroupState> state() {
+        return state;
+    }
+
     @Override
     public String toString() {
         return "(" +
             "groupId='" + groupId + '\'' +
             ", isSimpleConsumerGroup=" + isSimpleConsumerGroup +
+            ", state=" + state +
             ')';
     }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(groupId, isSimpleConsumerGroup, state);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+        ConsumerGroupListing other = (ConsumerGroupListing) obj;
+        if (groupId == null) {
+            if (other.groupId != null)
+                return false;
+        } else if (!groupId.equals(other.groupId))
+            return false;
+        if (isSimpleConsumerGroup != other.isSimpleConsumerGroup)
+            return false;
+        if (state == null) {
+            if (other.state != null)
+                return false;
+        } else if (!state.equals(other.state))
+            return false;
+        return true;
+    }
+
 }

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

@@ -3052,14 +3052,21 @@ void handleResponse(AbstractResponse abstractResponse) {
                     runnable.call(new Call("listConsumerGroups", deadline, new ConstantNodeIdProvider(node.id())) {
                         @Override
                         ListGroupsRequest.Builder createRequest(int timeoutMs) {
-                            return new ListGroupsRequest.Builder(new ListGroupsRequestData());
+                            List<String> states = options.states()
+                                    .stream()
+                                    .map(s -> s.toString())
+                                    .collect(Collectors.toList());
+                            return new ListGroupsRequest.Builder(new ListGroupsRequestData().setStatesFilter(states));
                         }
 
                         private void maybeAddConsumerGroup(ListGroupsResponseData.ListedGroup group) {
                             String protocolType = group.protocolType();
                             if (protocolType.equals(ConsumerProtocol.PROTOCOL_TYPE) || protocolType.isEmpty()) {
                                 final String groupId = group.groupId();
-                                final ConsumerGroupListing groupListing = new ConsumerGroupListing(groupId, protocolType.isEmpty());
+                                final Optional<ConsumerGroupState> state = group.groupState().equals("")
+                                        ? Optional.empty()
+                                        : Optional.of(ConsumerGroupState.parse(group.groupState()));
+                                final ConsumerGroupListing groupListing = new ConsumerGroupListing(groupId, protocolType.isEmpty(), state);
                                 results.addListing(groupListing);
                             }
                         }

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

@@ -17,6 +17,11 @@
 
 package org.apache.kafka.clients.admin;
 
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.kafka.common.ConsumerGroupState;
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
@@ -26,4 +31,23 @@
  */
 @InterfaceStability.Evolving
 public class ListConsumerGroupsOptions extends AbstractOptions<ListConsumerGroupsOptions> {
+
+    private Set<ConsumerGroupState> states = Collections.emptySet();
+
+    /**
+     * If states is set, only groups in these states will be returned by listConsumerGroups()
+     * Otherwise, all groups are returned.
+     * This operation is supported by brokers with version 2.6.0 or later.
+     */
+    public ListConsumerGroupsOptions inStates(Set<ConsumerGroupState> states) {
+        this.states = (states == null) ? Collections.emptySet() : new HashSet<>(states);
+        return this;
+    }
+
+    /**
+     * Returns the list of States that are requested or empty if no states have been specified
+     */
+    public Set<ConsumerGroupState> states() {
+        return states;
+    }
 }

clients/src/main/java/org/apache/kafka/common/ConsumerGroupState.java

@@ -45,7 +45,6 @@ public enum ConsumerGroupState {
         this.name = name;
     }
 
-
     /**
      * Parse a string into a consumer group state.
      */

clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.ListGroupsRequestData;
 import org.apache.kafka.common.message.ListGroupsResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -45,6 +46,10 @@ public Builder(ListGroupsRequestData data) {
 
         @Override
         public ListGroupsRequest build(short version) {
+            if (!data.statesFilter().isEmpty() && version < 4) {
+                throw new UnsupportedVersionException("The broker only supports ListGroups " +
+                        "v" + version + ", but we need v4 or newer to request groups by states.");
+            }
             return new ListGroupsRequest(data, version);
         }
 
@@ -66,6 +71,10 @@ public ListGroupsRequest(Struct struct, short version) {
         this.data = new ListGroupsRequestData(struct, version);
     }
 
+    public ListGroupsRequestData data() {
+        return data;
+    }
+
     @Override
     public ListGroupsResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         ListGroupsResponseData listGroupsResponseData = new ListGroupsResponseData().

clients/src/main/resources/common/message/ListGroupsRequest.json

@@ -20,8 +20,13 @@
   // Version 1 and 2 are the same as version 0.
   //
   // Version 3 is the first flexible version.
-  "validVersions": "0-3",
+  //
+  // Version 4 adds the StatesFilter field (KIP-518).
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
+    { "name": "StatesFilter", "type": "[]string", "versions": "4+",
+      "about": "The states of the groups we want to list. If empty all groups are returned with their state."
+    }
   ]
 }

clients/src/main/resources/common/message/ListGroupsResponse.json

@@ -22,7 +22,9 @@
   // Starting in version 2, on quota violation, brokers send out responses before throttling.
   //
   // Version 3 is the first flexible version.
-  "validVersions": "0-3",
+  //
+  // Version 4 adds the GroupState field (KIP-518).
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true,
@@ -34,7 +36,9 @@
       { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
         "about": "The group ID." },
       { "name": "ProtocolType", "type": "string", "versions": "0+",
-        "about": "The group protocol type." }
+        "about": "The group protocol type." },
+      { "name": "GroupState", "type": "string", "versions": "4+", "ignorable": true,
+        "about": "The group state name." }
     ]}
   ]
 }

clients/src/test/java/org/apache/kafka/clients/MockClient.java

@@ -215,15 +215,17 @@ public void send(ClientRequest request, long now) {
             AbstractRequest.Builder<?> builder = request.requestBuilder();
             short version = nodeApiVersions.latestUsableVersion(request.apiKey(), builder.oldestAllowedVersion(),
                     builder.latestAllowedVersion());
-            AbstractRequest abstractRequest = request.requestBuilder().build(version);
-            if (!futureResp.requestMatcher.matches(abstractRequest))
-                throw new IllegalStateException("Request matcher did not match next-in-line request " + abstractRequest + " with prepared response " + futureResp.responseBody);
 
             UnsupportedVersionException unsupportedVersionException = null;
-            if (futureResp.isUnsupportedRequest)
-                unsupportedVersionException = new UnsupportedVersionException("Api " +
-                        request.apiKey() + " with version " + version);
-
+            if (futureResp.isUnsupportedRequest) {
+                unsupportedVersionException = new UnsupportedVersionException(
+                        "Api " + request.apiKey() + " with version " + version);
+            } else {
+                AbstractRequest abstractRequest = request.requestBuilder().build(version);
+                if (!futureResp.requestMatcher.matches(abstractRequest))
+                    throw new IllegalStateException("Request matcher did not match next-in-line request "
+                            + abstractRequest + " with prepared response " + futureResp.responseBody);
+            }
             ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(),
                     request.createdTimeMs(), time.milliseconds(), futureResp.disconnected,
                     unsupportedVersionException, null, futureResp.responseBody);