[KIP-848 EA] Admin API for listing consumer groups now has an optional filter to return only groups of given types

Co-authored-by: mahajanadhitya <[email protected]>

Author: emasab

CHANGELOG.md

@@ -1,5 +1,16 @@
 # Confluent's Python client for Apache Kafka
 
+## v2.6.0
+
+v2.6.0 is a feature release with the following features, fixes and enhancements:
+
+- [KIP-848 EA](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol): Admin API for listing consumer groups now has an optional filter to return only groups of given types (#).
+
+confluent-kafka-python is based on librdkafka v2.6.0, see the
+[librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.6.0)
+for a complete list of changes, enhancements, fixes and upgrade considerations.
+
+
 ## v2.5.3
 
 v2.5.3 is a maintenance release with the following fixes and enhancements:
@@ -18,15 +29,15 @@ for a complete list of changes, enhancements, fixes and upgrade considerations.
 ## v2.5.0
 
 > [!WARNING]
-This version has introduced a regression in which an assert is triggered during **PushTelemetry** call. This happens when no metric is matched on the client side among those requested by broker subscription. 
+This version has introduced a regression in which an assert is triggered during **PushTelemetry** call. This happens when no metric is matched on the client side among those requested by broker subscription.
 >
 > You won't face any problem if:
 > * Broker doesn't support [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability).
 > * [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) feature is disabled on the broker side.
 > * [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) feature is disabled on the client side. This is enabled by default. Set configuration `enable.metrics.push` to `false`.
 > * If [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) is enabled on the broker side and there is no subscription configured there.
 > * If [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) is enabled on the broker side with subscriptions that match the [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) metrics defined on the client.
-> 
+>
 > Having said this, we strongly recommend using `v2.5.3` and above to not face this regression at all.
 
 v2.5.0 is a feature release with the following features, fixes and enhancements:

examples/adminapi.py

@@ -18,8 +18,9 @@
 # Example use of AdminClient operations.
 
 from confluent_kafka import (KafkaException, ConsumerGroupTopicPartitions,
-                             TopicPartition, ConsumerGroupState, TopicCollection,
-                             IsolationLevel, ElectionType)
+                             TopicPartition, ConsumerGroupState,
+                             TopicCollection, IsolationLevel,
+                             ConsumerGroupType, ElectionType)
 from confluent_kafka.admin import (AdminClient, NewTopic, NewPartitions, ConfigResource,
                                    ConfigEntry, ConfigSource, AclBinding,
                                    AclBindingFilter, ResourceType, ResourcePatternType,
@@ -471,18 +472,51 @@ def example_list(a, args):
                 print("id {} client_id: {} client_host: {}".format(m.id, m.client_id, m.client_host))
 
 
+def parse_list_consumer_groups_args(args, states, types):
+    def usage(message):
+        raise Exception(f"{message}\nUsage: list_consumer_groups [-states <state1> <state2> ..] "
+                        "[-types <type1> <type2> ..]")
+
+    if len(args) > 0:
+        typeArray = False
+        stateArray = False
+        lastArray = 0
+        for i in range(0, len(args)):
+            if (args[i] == "-states"):
+                if (stateArray):
+                    usage("Cannot pass the states flag (-states) more than once")
+                lastArray = 1
+                stateArray = True
+            elif (args[i] == "-types"):
+                if (typeArray):
+                    usage("Cannot pass the types flag (-types) more than once")
+                lastArray = 2
+                typeArray = True
+            else:
+                if (lastArray == 1):
+                    states.add(ConsumerGroupState[args[i]])
+                elif (lastArray == 2):
+                    types.add(ConsumerGroupType[args[i]])
+                else:
+                    usage(f"Unknown argument: {args[i]}")
+
+
 def example_list_consumer_groups(a, args):
     """
     List Consumer Groups
     """
-    states = {ConsumerGroupState[state] for state in args}
-    future = a.list_consumer_groups(request_timeout=10, states=states)
+
+    states = set()
+    types = set()
+    parse_list_consumer_groups_args(args, states, types)
+
+    future = a.list_consumer_groups(request_timeout=10, states=states, types=types)
     try:
         list_consumer_groups_result = future.result()
         print("{} consumer groups".format(len(list_consumer_groups_result.valid)))
         for valid in list_consumer_groups_result.valid:
-            print("    id: {} is_simple: {} state: {}".format(
-                valid.group_id, valid.is_simple_consumer_group, valid.state))
+            print("    id: {} is_simple: {} state: {} type: {}".format(
+                valid.group_id, valid.is_simple_consumer_group, valid.state, valid.type))
         print("{} errors".format(len(list_consumer_groups_result.errors)))
         for error in list_consumer_groups_result.errors:
             print("    error: {}".format(error))

src/confluent_kafka/__init__.py

@@ -22,6 +22,7 @@
 from ._model import (Node,  # noqa: F401
                      ConsumerGroupTopicPartitions,
                      ConsumerGroupState,
+                     ConsumerGroupType,
                      TopicCollection,
                      TopicPartitionInfo,
                      IsolationLevel,
@@ -49,7 +50,8 @@
            'Producer', 'DeserializingConsumer',
            'SerializingProducer', 'TIMESTAMP_CREATE_TIME', 'TIMESTAMP_LOG_APPEND_TIME',
            'TIMESTAMP_NOT_AVAILABLE', 'TopicPartition', 'Node',
-           'ConsumerGroupTopicPartitions', 'ConsumerGroupState', 'Uuid',
+           'ConsumerGroupTopicPartitions', 'ConsumerGroupState',
+           'ConsumerGroupType', 'Uuid',
            'IsolationLevel', 'TopicCollection', 'TopicPartitionInfo']
 
 __version__ = version()[0]

src/confluent_kafka/_model/__init__.py

@@ -95,6 +95,23 @@ def __lt__(self, other):
         return self.value < other.value
 
 
+class ConsumerGroupType(Enum):
+    """
+    Enumerates the different types of Consumer Group Type.
+    """
+    #: Type is not known or not set
+    UNKNOWN = cimpl.CONSUMER_GROUP_TYPE_UNKNOWN
+    #: Consumer Type
+    CONSUMER = cimpl.CONSUMER_GROUP_TYPE_CONSUMER
+    #: Classic Type
+    CLASSIC = cimpl.CONSUMER_GROUP_TYPE_CLASSIC
+
+    def __lt__(self, other):
+        if self.__class__ != other.__class__:
+            return NotImplemented
+        return self.value < other.value
+
+
 class TopicCollection:
     """
     Represents collection of topics in the form of different identifiers

src/confluent_kafka/admin/__init__.py

@@ -57,6 +57,7 @@
 from ._records import DeletedRecords  # noqa: F401
 
 from .._model import (TopicCollection as _TopicCollection,
+                      ConsumerGroupType as _ConsumerGroupType,
                       ElectionType as _ElectionType)
 
 from ..cimpl import (KafkaException,  # noqa: F401
@@ -898,6 +899,8 @@ def list_consumer_groups(self, **kwargs):
                   on broker, and response. Default: `socket.timeout.ms/1000.0`
         :param set(ConsumerGroupState) states: only list consumer groups which are currently in
                   these states.
+        :param set(ConsumerGroupType) types: only list consumer groups of
+                  these types.
 
         :returns: a future. Result method of the future returns :class:`ListConsumerGroupsResult`.
 
@@ -917,6 +920,16 @@ def list_consumer_groups(self, **kwargs):
                         raise TypeError("All elements of states must be of type ConsumerGroupState")
                 kwargs["states_int"] = [state.value for state in states]
             kwargs.pop("states")
+        if "types" in kwargs:
+            types = kwargs["types"]
+            if types is not None:
+                if not isinstance(types, set):
+                    raise TypeError("'types' must be a set")
+                for type in types:
+                    if not isinstance(type, _ConsumerGroupType):
+                        raise TypeError("All elements of types must be of type ConsumerGroupType")
+                kwargs["types_int"] = [type.value for type in types]
+            kwargs.pop("types")
 
         f, _ = AdminClient._make_futures([], None, AdminClient._make_list_consumer_groups_result)
 

src/confluent_kafka/admin/_group.py

@@ -14,7 +14,7 @@
 
 
 from .._util import ConversionUtil
-from .._model import ConsumerGroupState
+from .._model import ConsumerGroupState, ConsumerGroupType
 from ._acl import AclOperation
 
 
@@ -31,13 +31,17 @@ class ConsumerGroupListing:
         Whether a consumer group is simple or not.
     state : ConsumerGroupState
         Current state of the consumer group.
+    type : ConsumerGroupType
+        Type of the consumer group.
     """
 
-    def __init__(self, group_id, is_simple_consumer_group, state=None):
+    def __init__(self, group_id, is_simple_consumer_group, state=None, type=None):
         self.group_id = group_id
         self.is_simple_consumer_group = is_simple_consumer_group
         if state is not None:
             self.state = ConversionUtil.convert_to_enum(state, ConsumerGroupState)
+        if type is not None:
+            self.type = ConversionUtil.convert_to_enum(type, ConsumerGroupType)
 
 
 class ListConsumerGroupsResult: