[KIP-430] Added ACL Operations to Describe Topics, Cluster and ConsumerGroup APIs

CHANGELOG.md

@@ -1,5 +1,20 @@
 # Confluent's Python client for Apache Kafka
 
+## v2.3.0
+
+v2.3.0 is a feature release with the following features, fixes and enhancements:
+
+ * Add support for AdminAPI `DescribeCluster()` and `DescribeTopics()`. (@jainruchir, #1635)
+ * [KIP-430](https://cwiki.apache.org/confluence/display/KAFKA/KIP-430+-+Return+Authorized+Operations+in+Describe+Responses):
+   Return authorized operations in Describe Responses. (@jainruchir, #1635)
+ * Add `Rack` to the `Node` type, so AdminAPI calls can expose racks for brokers
+   (currently, all Describe Responses) (#1635, @jainruchir).
+
+confluent-kafka-python is based on librdkafka v2.3.0, see the
+[librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.3.0)
+for a complete list of changes, enhancements, fixes and upgrade considerations.
+
+
 ## v2.2.0
 
 v2.2.0 is a feature release with the following features, fixes and enhancements:

docs/index.rst

@@ -25,6 +25,11 @@ Supporting classes
     - :ref:`Message <pythonclient_message>`
     - :ref:`TopicPartition <pythonclient_topicpartition>`
     - :ref:`ThrottleEvent <pythonclient_throttleevent>`
+    - :ref:`TopicCollection <pythonclient_topic_collection>`
+    - :ref:`TopicPartitionInfo <pythonclient_topic_partition_info>`
+    - :ref:`Node <pythonclient_node>`
+    - :ref:`ConsumerGroupTopicPartitions <pythonclient_consumer_group_topic_partition>`
+    - :ref:`ConsumerGroupState <pythonclient_consumer_group_state>`
 
     - Errors:
        - :ref:`KafkaError <pythonclient_kafkaerror>`
@@ -55,6 +60,19 @@ Supporting classes
        - :ref:`UserScramCredentialAlteration <pythonclient_user_scram_credential_alteration>`
        - :ref:`UserScramCredentialUpsertion <pythonclient_user_scram_credential_upsertion>`
        - :ref:`UserScramCredentialDeletion <pythonclient_user_scram_credential_deletion>`
+       - :ref:`TopicDescription <pythonclient_topic_description>`
+       - :ref:`DescribeClusterResult <pythonclient_describe_cluster_result>`
+       - :ref:`BrokerMetadata <pythonclient_broker_metadata>`
+       - :ref:`ClusterMetadata <pythonclient_cluster_metadata>`
+       - :ref:`GroupMember <pythonclient_group_member>`
+       - :ref:`GroupMetadata <pythonclient_group_metadata>`
+       - :ref:`PartitionMetadata <pythonclient_partition_metadata>`
+       - :ref:`TopicMetadata <pythonclient_topic_metadata>`
+       - :ref:`ConsumerGroupListing <pythonclient_consumer_group_listing>`
+       - :ref:`ListConsumerGroupsResult <pythonclient_list_consumer_group_result>`
+       - :ref:`MemberAssignment <pythonclient_member_assignment>`
+       - :ref:`MemberDescription <pythonclient_member_description>`
+       - :ref:`ConsumerGroupDescription <pythonclient_consumer_group_description>`
 
 Experimental
    These classes are experimental and are likely to be removed, or subject to incompatible
@@ -84,6 +102,7 @@ AdminClient
 
 .. automodule:: confluent_kafka.admin
    :members:
+   :noindex:
 
 .. _pyclient_admin_newtopic:
 
@@ -247,6 +266,123 @@ UserScramCredentialDeletion
 .. autoclass:: confluent_kafka.admin.UserScramCredentialDeletion
    :members:
 
+.. _pythonclient_topic_description:
+
+****************
+TopicDescription
+****************
+
+.. autoclass:: confluent_kafka.admin.TopicDescription
+   :members:
+
+.. _pythonclient_describe_cluster_result:
+
+*********************
+DescribeClusterResult
+*********************
+
+.. autoclass:: confluent_kafka.admin.DescribeClusterResult
+   :members:
+
+.. _pythonclient_broker_metadata:
+
+**************
+BrokerMetadata
+**************
+
+.. autoclass:: confluent_kafka.admin.BrokerMetadata
+   :members:
+
+.. _pythonclient_cluster_metadata:
+
+***************
+ClusterMetadata
+***************
+
+.. autoclass:: confluent_kafka.admin.ClusterMetadata
+   :members:
+
+.. _pythonclient_group_member:
+
+***********
+GroupMember
+***********
+
+.. autoclass:: confluent_kafka.admin.GroupMember
+   :members:
+
+.. _pythonclient_group_metadata:
+
+*************
+GroupMetadata
+*************
+
+.. autoclass:: confluent_kafka.admin.GroupMetadata
+   :members:
+
+.. _pythonclient_partition_metadata:
+
+*****************
+PartitionMetadata
+*****************
+
+.. autoclass:: confluent_kafka.admin.PartitionMetadata
+   :members:
+
+.. _pythonclient_topic_metadata:
+
+*************
+TopicMetadata
+*************
+
+.. autoclass:: confluent_kafka.admin.TopicMetadata
+   :members:
+
+.. _pythonclient_consumer_group_listing:
+
+********************
+ConsumerGroupListing
+********************
+
+.. autoclass:: confluent_kafka.admin.ConsumerGroupListing
+   :members:
+
+.. _pythonclient_list_consumer_group_result:
+
+************************
+ListConsumerGroupsResult
+************************
+
+.. autoclass:: confluent_kafka.admin.ListConsumerGroupsResult
+   :members:
+
+.. _pythonclient_consumer_group_description:
+
+************************
+ConsumerGroupDescription
+************************
+
+.. autoclass:: confluent_kafka.admin.ConsumerGroupDescription
+   :members:
+
+.. _pythonclient_member_assignment:
+
+****************
+MemberAssignment
+****************
+
+.. autoclass:: confluent_kafka.admin.MemberAssignment
+   :members:
+
+.. _pythonclient_member_description:
+
+*****************
+MemberDescription
+*****************
+
+.. autoclass:: confluent_kafka.admin.MemberDescription
+   :members:
+
 .. _pythonclient_consumer:
 
 ********
@@ -255,6 +391,7 @@ Consumer
 
 .. autoclass:: confluent_kafka.Consumer
    :members:
+   :noindex:
 
 .. _serde_consumer:
 
@@ -275,6 +412,7 @@ Producer
 
 .. autoclass:: confluent_kafka.Producer
    :members:
+   :noindex:
 
 .. _serde_producer:
 
@@ -477,6 +615,51 @@ TopicPartition
 .. autoclass:: confluent_kafka.TopicPartition
    :members:
 
+.. _pythonclient_topic_collection:
+
+***************
+TopicCollection
+***************
+
+.. autoclass:: confluent_kafka.TopicCollection
+   :members:
+
+.. _pythonclient_topic_partition_info:
+
+******************
+TopicPartitionInfo
+******************
+
+.. autoclass:: confluent_kafka.TopicPartitionInfo
+   :members:
+
+.. _pythonclient_node:
+
+****
+Node
+****
+
+.. autoclass:: confluent_kafka.Node
+   :members:
+
+.. _pythonclient_consumer_group_topic_partition:
+
+****************************
+ConsumerGroupTopicPartitions
+****************************
+
+.. autoclass:: confluent_kafka.ConsumerGroupTopicPartitions
+   :members:
+
+.. _pythonclient_consumer_group_state:
+
+******************
+ConsumerGroupState
+******************
+
+.. autoclass:: confluent_kafka.ConsumerGroupState
+   :members:
+
 .. _serde_field:
 
 ************

examples/adminapi.py

@@ -18,7 +18,7 @@
 # Example use of AdminClient operations.
 
 from confluent_kafka import (KafkaException, ConsumerGroupTopicPartitions,
-                             TopicPartition, ConsumerGroupState)
+                             TopicPartition, ConsumerGroupState, TopicCollection)
 from confluent_kafka.admin import (AdminClient, NewTopic, NewPartitions, ConfigResource,
                                    ConfigEntry, ConfigSource, AclBinding,
                                    AclBindingFilter, ResourceType, ResourcePatternType,
@@ -492,8 +492,9 @@ def example_describe_consumer_groups(a, args):
     """
     Describe Consumer Groups
     """
-
-    futureMap = a.describe_consumer_groups(args, request_timeout=10)
+    include_auth_ops = bool(int(args[0]))
+    args = args[1:]
+    futureMap = a.describe_consumer_groups(args, include_authorized_operations=include_auth_ops, request_timeout=10)
 
     for group_id, future in futureMap.items():
         try:
@@ -502,7 +503,8 @@ def example_describe_consumer_groups(a, args):
             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(
+                f"  Coordinator        : {g.coordinator}")
             print("  Members: ")
             for member in g.members:
                 print("    Id                : {}".format(member.member_id))
@@ -513,12 +515,93 @@ def example_describe_consumer_groups(a, args):
                     print("    Assignments       :")
                     for toppar in member.assignment.topic_partitions:
                         print("      {} [{}]".format(toppar.topic, toppar.partition))
+            if (include_auth_ops):
+                print("  Authorized operations: ")
+                op_string = ""
+                for acl_op in g.authorized_operations:
+                    op_string += acl_op.name + "  "
+                print("    {}".format(op_string))
         except KafkaException as e:
             print("Error while describing group id '{}': {}".format(group_id, e))
         except Exception:
             raise
 
 
+def example_describe_topics(a, args):
+    """
+    Describe Topics
+    """
+    include_auth_ops = bool(int(args[0]))
+    args = args[1:]
+    topics = TopicCollection(topic_names=args)
+    futureMap = a.describe_topics(topics, request_timeout=10, include_authorized_operations=include_auth_ops)
+
+    for topic_name, future in futureMap.items():
+        try:
+            t = future.result()
+            print("Topic name             : {}".format(t.name))
+            if (t.is_internal):
+                print("Topic is Internal")
+
+            if (include_auth_ops):
+                print("Authorized operations  : ")
+                op_string = ""
+                for acl_op in t.authorized_operations:
+                    op_string += acl_op.name + "  "
+                print("    {}".format(op_string))
+
+            print("Partition Information")
+            for partition in t.partitions:
+                print("    Id                : {}".format(partition.id))
+                leader = partition.leader
+                print(f"    Leader            : {leader}")
+                print("    Replicas          : {}".format(len(partition.replicas)))
+                for replica in partition.replicas:
+                    print(f"         Replica            : {replica}")
+                print("    In-Sync Replicas  : {}".format(len(partition.isr)))
+                for isr in partition.isr:
+                    print(f"         In-Sync Replica    : {isr}")
+                print("")
+            print("")
+
+        except KafkaException as e:
+            print("Error while describing topic '{}': {}".format(topic_name, e))
+        except Exception:
+            raise
+
+
+def example_describe_cluster(a, args):
+    """
+    Describe Cluster
+    """
+    include_auth_ops = bool(int(args[0]))
+    args = args[1:]
+    future = a.describe_cluster(request_timeout=10, include_authorized_operations=include_auth_ops)
+    try:
+        c = future.result()
+        print("Cluster_id           : {}".format(c.cluster_id))
+
+        if (c.controller):
+            print(f"Controller: {c.controller}")
+        else:
+            print("No Controller Information Available")
+
+        print("Nodes                :")
+        for node in c.nodes:
+            print(f"  Node: {node}")
+
+        if (include_auth_ops):
+            print("Authorized operations: ")
+            op_string = ""
+            for acl_op in c.authorized_operations:
+                op_string += acl_op.name + "  "
+            print("    {}".format(op_string))
+    except KafkaException as e:
+        print("Error while describing cluster: {}".format(e))
+    except Exception:
+        raise
+
+
 def example_delete_consumer_groups(a, args):
     """
     Delete Consumer Groups
@@ -704,7 +787,9 @@ def example_alter_user_scram_credentials(a, args):
                          '<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(' describe_consumer_groups <include_authorized_operations> <group1> <group2> ..\n')
+        sys.stderr.write(' describe_topics <include_authorized_operations> <topic1> <topic2> ..\n')
+        sys.stderr.write(' describe_cluster <include_authorized_operations>\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(
@@ -737,6 +822,8 @@ def example_alter_user_scram_credentials(a, args):
               'list': example_list,
               'list_consumer_groups': example_list_consumer_groups,
               'describe_consumer_groups': example_describe_consumer_groups,
+              'describe_topics': example_describe_topics,
+              'describe_cluster': example_describe_cluster,
               '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,