KAFKA-17897 Deprecate Admin.listConsumerGroups (#19477)

The final part of KIP-1043 is to deprecate Admin.listConsumerGroups() in
favour of Admin.listGroups() which works for all group types.

Reviewers: PoAn Yang <[email protected]>, Chia-Ping Tsai
<[email protected]>

Author: AndrewJSchofield

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

@@ -879,20 +879,24 @@ default DescribeConsumerGroupsResult describeConsumerGroups(Collection<String> g
 
     /**
      * List the consumer groups available in the cluster.
+     * @deprecated Since 4.1. Use {@link Admin#listGroups(ListGroupsOptions)} instead.
      *
      * @param options The options to use when listing the consumer groups.
      * @return The ListConsumerGroupsResult.
      */
+    @Deprecated(since = "4.1", forRemoval = true)
     ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options);
 
     /**
      * List the consumer groups available in the cluster with the default options.
      * <p>
      * This is a convenience method for {@link #listConsumerGroups(ListConsumerGroupsOptions)} with default options.
      * See the overload for more details.
+     * @deprecated Since 4.1. Use {@link Admin#listGroups(ListGroupsOptions)} instead.
      *
      * @return The ListConsumerGroupsResult.
      */
+    @Deprecated(since = "4.1", forRemoval = true)
     default ListConsumerGroupsResult listConsumerGroups() {
         return listConsumerGroups(new ListConsumerGroupsOptions());
     }

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

@@ -26,7 +26,9 @@
 
 /**
  * A listing of a consumer group in the cluster.
+ * @deprecated Since 4.1. Use {@link Admin#listGroups(ListGroupsOptions)} and {@link GroupListing} instead.
  */
+@Deprecated(since = "4.1")
 public class ConsumerGroupListing {
     private final String groupId;
     private final boolean isSimpleConsumerGroup;

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

@@ -159,6 +159,8 @@ public DescribeConsumerGroupsResult describeConsumerGroups(Collection<String> gr
     }
 
     @Override
+    @Deprecated
+    @SuppressWarnings("removal")
     public ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options) {
         return delegate.listConsumerGroups(options);
     }

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

@@ -3614,6 +3614,7 @@ public DescribeConsumerGroupsResult describeConsumerGroups(final Collection<Stri
                 .collect(Collectors.toMap(entry -> entry.getKey().idValue, Map.Entry::getValue)));
     }
 
+    @Deprecated
     private static final class ListConsumerGroupsResults {
         private final List<Throwable> errors;
         private final HashMap<String, ConsumerGroupListing> listings;
@@ -3657,6 +3658,8 @@ private synchronized void tryComplete() {
     }
 
     @Override
+    @SuppressWarnings("removal")
+    @Deprecated(since = "4.1", forRemoval = true)
     public ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options) {
         final KafkaFutureImpl<Collection<Object>> all = new KafkaFutureImpl<>();
         final long nowMetadata = time.milliseconds();

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

@@ -28,7 +28,10 @@
 
 /**
  * Options for {@link Admin#listConsumerGroups()}.
+ * @deprecated Since 4.1. Use {@link Admin#listGroups(ListGroupsOptions)} instead.
  */
+@Deprecated(since = "4.1")
+@SuppressWarnings("removal")
 public class ListConsumerGroupsOptions extends AbstractOptions<ListConsumerGroupsOptions> {
 
     private Set<GroupState> groupStates = Collections.emptySet();

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

@@ -25,8 +25,10 @@
 
 /**
  * The result of the {@link Admin#listConsumerGroups()} call.
- * <p>
+ * @deprecated Since 4.1. Use {@link Admin#listGroups(ListGroupsOptions)} instead.
  */
+@Deprecated(since = "4.1")
+@SuppressWarnings("removal")
 public class ListConsumerGroupsResult {
     private final KafkaFutureImpl<Collection<ConsumerGroupListing>> all;
     private final KafkaFutureImpl<Collection<ConsumerGroupListing>> valid;

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

@@ -17,11 +17,11 @@
 
 package org.apache.kafka.clients.admin;
 
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
 import org.apache.kafka.common.GroupState;
 import org.apache.kafka.common.GroupType;
 import org.apache.kafka.common.annotation.InterfaceStability;
 
-import java.util.Collections;
 import java.util.Set;
 
 /**
@@ -32,16 +32,32 @@
 @InterfaceStability.Evolving
 public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {
 
-    private Set<GroupState> groupStates = Collections.emptySet();
-    private Set<GroupType> types = Collections.emptySet();
+    private Set<GroupState> groupStates = Set.of();
+    private Set<GroupType> types = Set.of();
+    private Set<String> protocolTypes = Set.of();
+
+    /**
+     * Only consumer groups will be returned by listGroups().
+     * This operation sets filters on group type and protocol type which select consumer groups.
+     */
+    public static ListGroupsOptions forConsumerGroups() {
+        return new ListGroupsOptions()
+            .withTypes(Set.of(GroupType.CLASSIC, GroupType.CONSUMER))
+            .withProtocolTypes(Set.of("", ConsumerProtocol.PROTOCOL_TYPE));
+    }
 
     /**
      * If groupStates is set, only groups in these states will be returned by listGroups().
      * Otherwise, all groups are returned.
      * This operation is supported by brokers with version 2.6.0 or later.
      */
     public ListGroupsOptions inGroupStates(Set<GroupState> groupStates) {
-        this.groupStates = (groupStates == null || groupStates.isEmpty()) ? Collections.emptySet() : Set.copyOf(groupStates);
+        this.groupStates = (groupStates == null || groupStates.isEmpty()) ? Set.of() : Set.copyOf(groupStates);
+        return this;
+    }
+
+    public ListGroupsOptions withProtocolTypes(Set<String> protocolTypes) {
+        this.protocolTypes = (protocolTypes == null || protocolTypes.isEmpty()) ? Set.of() : Set.copyOf(protocolTypes);
         return this;
     }
 
@@ -61,6 +77,13 @@ public Set<GroupState> groupStates() {
         return groupStates;
     }
 
+    /**
+     * Returns the list of protocol types that are requested or empty if no protocol types have been specified.
+     */
+    public Set<String> protocolTypes() {
+        return protocolTypes;
+    }
+
     /**
      * Returns the list of group types that are requested or empty if no types have been specified.
      */

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -3268,6 +3268,7 @@ public void testDescribeClusterHandleUnsupportedVersionForIncludingFencedBrokers
     }
 
     @Test
+    @SuppressWarnings("removal")
     public void testListConsumerGroups() throws Exception {
         try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(4, 0),
                 AdminClientConfig.RETRIES_CONFIG, "2")) {
@@ -3377,6 +3378,7 @@ public void testListConsumerGroups() throws Exception {
     }
 
     @Test
+    @SuppressWarnings("removal")
     public void testListConsumerGroupsMetadataFailure() throws Exception {
         final Cluster cluster = mockCluster(3, 0);
         final Time time = new MockTime();
@@ -3400,6 +3402,7 @@ public void testListConsumerGroupsMetadataFailure() throws Exception {
     }
 
     @Test
+    @SuppressWarnings("removal")
     public void testListConsumerGroupsWithStates() throws Exception {
         try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
             env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
@@ -3433,6 +3436,7 @@ public void testListConsumerGroupsWithStates() throws Exception {
     }
 
     @Test
+    @SuppressWarnings("removal")
     public void testListConsumerGroupsWithTypes() throws Exception {
         try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
             env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
@@ -3495,6 +3499,7 @@ public void testListConsumerGroupsWithTypes() throws Exception {
     }
 
     @Test
+    @SuppressWarnings("removal")
     public void testListConsumerGroupsWithStatesOlderBrokerVersion() throws Exception {
         ApiVersion listGroupV3 = new ApiVersion()
                 .setApiKey(ApiKeys.LIST_GROUPS.id)
@@ -3533,6 +3538,7 @@ public void testListConsumerGroupsWithStatesOlderBrokerVersion() throws Exceptio
     }
 
     @Test
+    @SuppressWarnings("removal")
     public void testListConsumerGroupsWithTypesOlderBrokerVersion() throws Exception {
         ApiVersion listGroupV4 = new ApiVersion()
             .setApiKey(ApiKeys.LIST_GROUPS.id)

clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java

@@ -735,6 +735,7 @@ public synchronized DescribeConsumerGroupsResult describeConsumerGroups(Collecti
     }
 
     @Override
+    @SuppressWarnings("removal")
     public synchronized ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options) {
         KafkaFutureImpl<Collection<Object>> future = new KafkaFutureImpl<>();
         future.complete(groupConfigs.keySet().stream().map(g -> new ConsumerGroupListing(g, false)).collect(Collectors.toList()));

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointConnector.java

@@ -17,8 +17,9 @@
 package org.apache.kafka.connect.mirror;
 
 import org.apache.kafka.clients.admin.Admin;
-import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.clients.admin.GroupListing;
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
+import org.apache.kafka.clients.admin.ListGroupsOptions;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.Config;
@@ -225,7 +226,7 @@ private void loadInitialConsumerGroups()
     Set<String> findConsumerGroups()
             throws InterruptedException, ExecutionException {
         List<String> filteredGroups = listConsumerGroups().stream()
-                .map(ConsumerGroupListing::groupId)
+                .map(GroupListing::groupId)
                 .filter(this::shouldReplicateByGroupFilter)
                 .collect(Collectors.toList());
 
@@ -252,10 +253,10 @@ Set<String> findConsumerGroups()
         return checkpointGroups;
     }
 
-    Collection<ConsumerGroupListing> listConsumerGroups()
+    Collection<GroupListing> listConsumerGroups()
             throws InterruptedException, ExecutionException {
         return adminCall(
-                () -> sourceAdminClient.listConsumerGroups().valid().get(),
+                () -> sourceAdminClient.listGroups(ListGroupsOptions.forConsumerGroups()).valid().get(),
                 () -> "list consumer groups on " + config.sourceClusterAlias() + " cluster"
         );
     }