Implemented KIP-88, KIP-222, KIP-518 and partially KIP-396

CHANGELOG.md

@@ -4,9 +4,29 @@
 ## Next Version
 
  - Added metadata to `TopicPartition` type and `commit()` (#1410).
- - Added `consumer.memberid()` for getting member id assigned to 
+ - Added `consumer.memberid()` for getting member id assigned to
    the consumer in a consumer group (#1154).
- - Added Python 3.11 wheels
+ - Implemented `nb_bool` method for the Producer, so that the default (which uses len)
+   will not be used. This avoids situations where producers with no enqueued items would
+   evaluate to False (#1445).
+ - Added Python 3.11 wheels.
+ - [KIP-222](https://cwiki.apache.org/confluence/display/KAFKA/KIP-222+-+Add+Consumer+Group+operations+to+Admin+API)
+   Add Consumer Group operations to Admin API.
+ - [KIP-518](https://cwiki.apache.org/confluence/display/KAFKA/KIP-518%3A+Allow+listing+consumer+groups+per+state)
+   Allow listing consumer groups per state.
+ - [KIP-396](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=97551484)
+   Partially implemented: support for AlterConsumerGroupOffsets.
+ - As result of the above KIPs, added (#1449)
+   - `list_consumer_groups` Admin operation. Supports listing by state.
+   - `describe_consumer_groups` Admin operation. Supports multiple groups.
+   - `delete_consumer_groups` Admin operation. Supports multiple groups.
+   - `list_consumer_group_offsets` Admin operation. Currently, only supports 1 group with multiple partitions. Supports require_stable option.
+   - `alter_consumer_group_offsets` Admin operation. Currently, only supports 1 group with multiple offsets.
+ - Added `normalize.schemas` configuration property to Schema Registry client
+
+confluent-kafka-python is based on librdkafka v2.0.0, see the
+[librdkafka release notes](https://github.com/edenhill/librdkafka/releases/tag/v2.0.0)
+for a complete list of changes, enhancements, fixes and upgrade considerations.
 
 
 ## v1.9.2

examples/adminapi.py

@@ -17,10 +17,11 @@
 
 # Example use of AdminClient operations.
 
+from confluent_kafka import (KafkaException, ConsumerGroupTopicPartitions,
+                             TopicPartition)
 from confluent_kafka.admin import (AdminClient, NewTopic, NewPartitions, ConfigResource, ConfigSource,
-                                   AclBinding, AclBindingFilter, ResourceType, ResourcePatternType,
-                                   AclOperation, AclPermissionType)
-from confluent_kafka import KafkaException
+                                   AclBinding, AclBindingFilter, ResourceType, ResourcePatternType, AclOperation,
+                                   AclPermissionType, ConsumerGroupState)
 import sys
 import threading
 import logging
@@ -419,17 +420,146 @@ def example_list(a, args):
         print(" {} consumer groups".format(len(groups)))
         for g in groups:
             if g.error is not None:
-                errstr = ": {}".format(t.error)
+                errstr = ": {}".format(g.error)
             else:
                 errstr = ""
 
             print(" \"{}\" with {} member(s), protocol: {}, protocol_type: {}{}".format(
-                  g, len(g.members), g.protocol, g.protocol_type, errstr))
+                g, len(g.members), g.protocol, g.protocol_type, errstr))
 
             for m in g.members:
                 print("id {} client_id: {} client_host: {}".format(m.id, m.client_id, m.client_host))
 
 
+def example_list_consumer_groups(a, args):
+    """
+    List Consumer Groups
+    """
+    states = [ConsumerGroupState[state] for state in args]
+    future = a.list_consumer_groups(timeout=10, states=states)
+    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("{} errors".format(len(list_consumer_groups_result.errors)))
+        for error in list_consumer_groups_result.errors:
+            print("    error: {}".format(error))
+    except Exception:
+        raise
+
+
+def example_describe_consumer_groups(a, args):
+    """
+    Describe Consumer Groups
+    """
+
+    futureMap = a.describe_consumer_groups(args, timeout=10)
+
+    for group_id, future in futureMap.items():
+        try:
+            g = future.result()
+            print("Group Id: {}".format(g.group_id))
+            print("  Is Simple          : {}".format(g.is_simple_consumer_group))
+            print("  State              : {}".format(g.state))
+            print("  Partition Assignor : {}".format(g.partition_assignor))
+            print("  Coordinator        : ({}) {}:{}".format(g.coordinator.id, g.coordinator.host, g.coordinator.port))
+            print("  Members: ")
+            for member in g.members:
+                print("    Id                : {}".format(member.member_id))
+                print("    Host              : {}".format(member.host))
+                print("    Client Id         : {}".format(member.client_id))
+                print("    Group Instance Id : {}".format(member.group_instance_id))
+                if member.assignment:
+                    print("    Assignments       :")
+                    for toppar in member.assignment.topic_partitions:
+                        print("      {} [{}]".format(toppar.topic, toppar.partition))
+        except KafkaException as e:
+            print("Error while describing group id '{}': {}".format(group_id, e))
+        except Exception:
+            raise
+
+
+def example_delete_consumer_groups(a, args):
+    """
+    Delete Consumer Groups
+    """
+    groups = a.delete_consumer_groups(args, timeout=10)
+    for group_id, future in groups.items():
+        try:
+            response = future.result()
+            print("Deleted group with id '" + response.group_id + "' successfully")
+        except KafkaException as e:
+            print("Error deleting group id '{}': {}".format(group_id, e))
+        except Exception:
+            raise
+
+
+def example_list_consumer_group_offsets(a, args):
+    """
+    List consumer group offsets
+    """
+
+    topic_partitions = []
+    for topic, partition in zip(args[1::2], args[2::2]):
+        topic_partitions.append(TopicPartition(topic, int(partition)))
+    if len(topic_partitions) == 0:
+        topic_partitions = None
+    groups = [ConsumerGroupTopicPartitions(args[0], topic_partitions)]
+
+    futureMap = a.list_consumer_group_offsets(groups)
+
+    for request, future in futureMap.items():
+        try:
+            response_offset_info = future.result()
+            print("Group: " + response_offset_info.group_id)
+            for topic_partition in response_offset_info.topic_partitions:
+                if topic_partition.error:
+                    print("    Error: " + topic_partition.error.str() + " occurred with " +
+                          topic_partition.topic + " [" + str(topic_partition.partition) + "]")
+                else:
+                    print("    " + topic_partition.topic +
+                          " [" + str(topic_partition.partition) + "]: " + str(topic_partition.offset))
+
+        except KafkaException as e:
+            print("Failed to describe {}: {}".format(request.group_id, e))
+        except Exception:
+            raise
+
+
+def example_alter_consumer_group_offsets(a, args):
+    """
+    Alter consumer group offsets
+    """
+
+    topic_partitions = []
+    for topic, partition, offset in zip(args[1::3], args[2::3], args[3::3]):
+        topic_partitions.append(TopicPartition(topic, int(partition), int(offset)))
+    if len(topic_partitions) == 0:
+        topic_partitions = None
+    groups = [ConsumerGroupTopicPartitions(args[0], topic_partitions)]
+
+    futureMap = a.alter_consumer_group_offsets(groups)
+
+    for request, future in futureMap.items():
+        try:
+            response_offset_info = future.result()
+            print("Group: " + response_offset_info.group_id)
+            for topic_partition in response_offset_info.topic_partitions:
+                if topic_partition.error:
+                    print("    Error: " + topic_partition.error.str() + " occurred with " +
+                          topic_partition.topic + " [" + str(topic_partition.partition) + "]")
+                else:
+                    print("    " + topic_partition.topic +
+                          " [" + str(topic_partition.partition) + "]: " + str(topic_partition.offset))
+
+        except KafkaException as e:
+            print("Failed to describe {}: {}".format(request.group_id, e))
+        except Exception:
+            raise
+
+
 if __name__ == '__main__':
     if len(sys.argv) < 3:
         sys.stderr.write('Usage: %s <bootstrap-brokers> <operation> <args..>\n\n' % sys.argv[0])
@@ -449,6 +579,14 @@ def example_list(a, args):
         sys.stderr.write(' delete_acls <resource_type1> <resource_name1> <resource_patter_type1> ' +
                          '<principal1> <host1> <operation1> <permission_type1> ..\n')
         sys.stderr.write(' list [<all|topics|brokers|groups>]\n')
+        sys.stderr.write(' list_consumer_groups [<state1> <state2> ..]\n')
+        sys.stderr.write(' describe_consumer_groups <group1> <group2> ..\n')
+        sys.stderr.write(' delete_consumer_groups <group1> <group2> ..\n')
+        sys.stderr.write(' list_consumer_group_offsets <group> [<topic1> <partition1> <topic2> <partition2> ..]\n')
+        sys.stderr.write(
+            ' alter_consumer_group_offsets <group> <topic1> <partition1> <offset1> ' +
+            '<topic2> <partition2> <offset2> ..\n')
+
         sys.exit(1)
 
     broker = sys.argv[1]
@@ -467,7 +605,12 @@ def example_list(a, args):
               'create_acls': example_create_acls,
               'describe_acls': example_describe_acls,
               'delete_acls': example_delete_acls,
-              'list': example_list}
+              'list': example_list,
+              'list_consumer_groups': example_list_consumer_groups,
+              'describe_consumer_groups': example_describe_consumer_groups,
+              'delete_consumer_groups': example_delete_consumer_groups,
+              'list_consumer_group_offsets': example_list_consumer_group_offsets,
+              'alter_consumer_group_offsets': example_alter_consumer_group_offsets}
 
     if operation not in opsmap:
         sys.stderr.write('Unknown operation: %s\n' % operation)

src/confluent_kafka/__init__.py

@@ -19,6 +19,7 @@
 from .deserializing_consumer import DeserializingConsumer
 from .serializing_producer import SerializingProducer
 from .error import KafkaException, KafkaError
+from ._model import Node, ConsumerGroupTopicPartitions
 
 from .cimpl import (Producer,
                     Consumer,
@@ -40,7 +41,8 @@
            'OFFSET_BEGINNING', 'OFFSET_END', 'OFFSET_INVALID', 'OFFSET_STORED',
            'Producer', 'DeserializingConsumer',
            'SerializingProducer', 'TIMESTAMP_CREATE_TIME', 'TIMESTAMP_LOG_APPEND_TIME',
-           'TIMESTAMP_NOT_AVAILABLE', 'TopicPartition']
+           'TIMESTAMP_NOT_AVAILABLE', 'TopicPartition', 'Node',
+           'ConsumerGroupTopicPartitions']
 
 __version__ = version()[0]
 

src/confluent_kafka/_model/__init__.py

@@ -0,0 +1,57 @@
+# Copyright 2022 Confluent Inc.
+#
+# Licensed 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.
+
+
+class Node:
+    """
+    Represents node information.
+    Used by :class:`ConsumerGroupDescription`
+
+    Parameters
+    ----------
+    id: int
+        The node id of this node.
+    id_string:
+        String representation of the node id.
+    host:
+        The host name for this node.
+    port: int
+        The port for this node.
+    rack: str
+        The rack for this node.
+    """
+    def __init__(self, id, host, port, rack=None):
+        self.id = id
+        self.id_string = str(id)
+        self.host = host
+        self.port = port
+        self.rack = rack
+
+
+class ConsumerGroupTopicPartitions:
+    """
+    Represents consumer group and its topic partition information.
+    Used by :meth:`AdminClient.list_consumer_group_offsets` and
+    :meth:`AdminClient.alter_consumer_group_offsets`.
+
+    Parameters
+    ----------
+    group_id: str
+        Id of the consumer group.
+    topic_partitions : list(TopicPartition)
+        List of topic partitions information.
+    """
+    def __init__(self, group_id, topic_partitions=None):
+        self.group_id = group_id
+        self.topic_partitions = topic_partitions

src/confluent_kafka/_util/__init__.py

@@ -0,0 +1,16 @@
+# Copyright 2022 Confluent Inc.
+#
+# Licensed 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.
+
+from .validation_util import ValidationUtil  # noqa: F401
+from .conversion_util import ConversionUtil  # noqa: F401

src/confluent_kafka/_util/conversion_util.py

@@ -0,0 +1,38 @@
+# Copyright 2022 Confluent Inc.
+#
+# Licensed 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.
+
+from enum import Enum
+
+
+class ConversionUtil:
+    @staticmethod
+    def convert_to_enum(val, enum_clazz):
+        if type(enum_clazz) is not type(Enum):
+            raise TypeError("'enum_clazz' must be of type Enum")
+
+        if type(val) == str:
+            # Allow it to be specified as case-insensitive string, for convenience.
+            try:
+                val = enum_clazz[val.upper()]
+            except KeyError:
+                raise ValueError("Unknown value \"%s\": should be a %s" % (val, enum_clazz.__name__))
+
+        elif type(val) == int:
+            # The C-code passes restype as an int, convert to enum.
+            val = enum_clazz(val)
+
+        elif type(val) != enum_clazz:
+            raise TypeError("Unknown value \"%s\": should be a %s" % (val, enum_clazz.__name__))
+
+        return val