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,22 +17,30 @@
 
 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.
      *
      * @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) {
+    public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup, Optional<ConsumerGroupState> state) {
         this.groupId = groupId;
         this.isSimpleConsumerGroup = isSimpleConsumerGroup;
+        this.state = state;
     }
 
     /**
@@ -49,11 +57,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,23 @@ 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().isEmpty())
+                                    ? null
+                                    : 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() == null
+                                        ? 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,30 @@
  */
 @InterfaceStability.Evolving
 public class ListConsumerGroupsOptions extends AbstractOptions<ListConsumerGroupsOptions> {
+
+    private Set<ConsumerGroupState> states = Collections.emptySet();
+
+    /**
+     * Only groups in these states will be returned by listConsumerGroups()
+     * If not set, all groups are returned with their states
+     */
+    public ListConsumerGroupsOptions inStates(Set<ConsumerGroupState> states) {
+        this.states = (states == null) ? Collections.emptySet() : new HashSet<>(states);
+        return this;
+    }
+
+    /**
+     * All groups with their states will be returned by listConsumerGroups()
+     */
+    public ListConsumerGroupsOptions inAnyState() {
+        this.states = Collections.emptySet();
+        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() != null && 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,14 @@
   // 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+", "nullableVersions": "4+", "default": "null",
+      "about": "The states of the groups we want to list. If empty or null, all groups are returned with their state.", "fields": [
+      { "name": "Name", "type": "string", "versions": "4+", "about": "The name of the group 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+", "nullableVersions": "0+", "ignorable": true, "default": "null",
+        "about": "The group state name." }
     ]}
   ]
 }

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

@@ -215,15 +215,22 @@ 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);
-
+                unsupportedVersionException = new UnsupportedVersionException(
+                        "Api " + request.apiKey() + " with version " + version);
+            try {
+                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);
+
+            } catch (UnsupportedVersionException uve) {
+                if (unsupportedVersionException == null) {
+                    throw uve;
+                }
+            }
             ClientResponse resp = new ClientResponse(request.makeHeader(version), request.callback(), request.destination(),
                     request.createdTimeMs(), time.milliseconds(), futureResp.disconnected,
                     unsupportedVersionException, null, futureResp.responseBody);