Rename KafkaAdmin to KafkaAdminClient

docs/apidoc/KafkaAdminClient.rst

@@ -0,0 +1,5 @@
+KafkaAdminClient
+===========
+
+.. autoclass:: kafka.admin.KafkaAdminClient
+    :members:

docs/apidoc/modules.rst

@@ -5,7 +5,7 @@ kafka-python API
 
    KafkaConsumer
    KafkaProducer
-   KafkaAdmin
+   KafkaAdminClient
    KafkaClient
    BrokerConnection
    ClusterMetadata

kafka/__init__.py

@@ -18,7 +18,7 @@ def emit(self, record):
 logging.getLogger(__name__).addHandler(NullHandler())
 
 
-from kafka.admin import KafkaAdmin
+from kafka.admin import KafkaAdminClient
 from kafka.consumer import KafkaConsumer
 from kafka.consumer.subscription_state import ConsumerRebalanceListener
 from kafka.producer import KafkaProducer
@@ -47,7 +47,7 @@ def __init__(self, *args, **kwargs):
 
 
 __all__ = [
-    'KafkaAdmin',
+    'KafkaAdminClient',
     'KafkaConsumer', 'KafkaProducer', 'KafkaClient', 'BrokerConnection',
     'SimpleClient', 'SimpleProducer', 'KeyedProducer',
     'RoundRobinPartitioner', 'HashedPartitioner',

kafka/admin/__init__.py

@@ -1,10 +1,10 @@
 from __future__ import absolute_import
 
 from kafka.admin.config_resource import ConfigResource, ConfigResourceType
-from kafka.admin.kafka import KafkaAdmin
+from kafka.admin.client import KafkaAdminClient
 from kafka.admin.new_topic import NewTopic
 from kafka.admin.new_partitions import NewPartitions
 
 __all__ = [
-    'ConfigResource', 'ConfigResourceType', 'KafkaAdmin', 'NewTopic', 'NewPartitions'
+    'ConfigResource', 'ConfigResourceType', 'KafkaAdminClient', 'NewTopic', 'NewPartitions'
 ]

kafka/admin/kafka.py → kafka/admin/client.py

@@ -25,7 +25,7 @@
 log = logging.getLogger(__name__)
 
 
-class KafkaAdmin(object):
+class KafkaAdminClient(object):
     """A class for administering the Kafka cluster.
 
     Warning:
@@ -35,7 +35,7 @@ class KafkaAdmin(object):
         nicer, more pythonic objects. Unfortunately, this will likely break
         those interfaces.
 
-    The KafkaAdmin class will negotiate for the latest version of each message
+    The KafkaAdminClient class will negotiate for the latest version of each message
     protocol format supported by both the kafka-python client library and the
     Kafka broker. Usage of optional fields from protocol versions that are not
     supported by the broker will result in IncompatibleBrokerVersion exceptions.
@@ -174,7 +174,7 @@ class KafkaAdmin(object):
     }
 
     def __init__(self, **configs):
-        log.debug("Starting KafkaAdmin interface.")
+        log.debug("Starting KafkaAdminClient with configuration: %s", configs)
         extra_configs = set(configs).difference(self.DEFAULT_CONFIG)
         if extra_configs:
             raise KafkaConfigurationError("Unrecognized configs: {}".format(extra_configs))
@@ -200,18 +200,18 @@ def __init__(self, **configs):
 
         self._closed = False
         self._refresh_controller_id()
-        log.debug("KafkaAdmin interface started.")
+        log.debug("KafkaAdminClient started.")
 
     def close(self):
-        """Close the KafkaAdmin connection to the Kafka broker."""
+        """Close the KafkaAdminClient connection to the Kafka broker."""
         if not hasattr(self, '_closed') or self._closed:
-            log.info("KafkaAdmin interface already closed.")
+            log.info("KafkaAdminClient already closed.")
             return
 
         self._metrics.close()
         self._client.close()
         self._closed = True
-        log.debug("KafkaAdmin interface has closed.")
+        log.debug("KafkaAdminClient is now closed.")
 
     def _matching_api_version(self, operation):
         """Find the latest version of the protocol operation supported by both
@@ -252,7 +252,7 @@ def _refresh_controller_id(self):
             controller_version = self._client.check_version(controller_id)
             if controller_version < (0, 10, 0):
                 raise IncompatibleBrokerVersion(
-                    "The controller appears to be running Kafka {}. KafkaAdmin requires brokers >= 0.10.0.0."
+                    "The controller appears to be running Kafka {}. KafkaAdminClient requires brokers >= 0.10.0.0."
                     .format(controller_version))
             self._controller_id = controller_id
         else:
@@ -391,7 +391,7 @@ def create_topics(self, new_topics, timeout_ms=None, validate_only=False):
             )
         else:
             raise NotImplementedError(
-                "Support for CreateTopics v{} has not yet been added to KafkaAdmin."
+                "Support for CreateTopics v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         # TODO convert structs to a more pythonic interface
         # TODO raise exceptions if errors
@@ -415,7 +415,7 @@ def delete_topics(self, topics, timeout_ms=None):
             response = self._send_request_to_controller(request)
         else:
             raise NotImplementedError(
-                "Support for DeleteTopics v{} has not yet been added to KafkaAdmin."
+                "Support for DeleteTopics v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         return response
 
@@ -474,7 +474,7 @@ def describe_configs(self, config_resources, include_synonyms=False):
             )
         else:
             raise NotImplementedError(
-                "Support for DescribeConfigs v{} has not yet been added to KafkaAdmin."
+                "Support for DescribeConfigs v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         return self._send_request_to_node(self._client.least_loaded_node(), request)
 
@@ -507,7 +507,7 @@ def alter_configs(self, config_resources):
             )
         else:
             raise NotImplementedError(
-                "Support for AlterConfigs v{} has not yet been added to KafkaAdmin."
+                "Support for AlterConfigs v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         # TODO the Java client has the note:
         # // We must make a separate AlterConfigs request for every BROKER resource we want to alter
@@ -553,7 +553,7 @@ def create_partitions(self, topic_partitions, timeout_ms=None, validate_only=Fal
             )
         else:
             raise NotImplementedError(
-                "Support for CreatePartitions v{} has not yet been added to KafkaAdmin."
+                "Support for CreatePartitions v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         return self._send_request_to_controller(request)
 
@@ -625,7 +625,7 @@ def describe_consumer_groups(self, group_ids, group_coordinator_id=None):
                 group_descriptions.append(group_description)
             else:
                 raise NotImplementedError(
-                    "Support for DescribeGroups v{} has not yet been added to KafkaAdmin."
+                    "Support for DescribeGroups v{} has not yet been added to KafkaAdminClient."
                     .format(version))
         return group_descriptions
 
@@ -674,7 +674,7 @@ def list_consumer_groups(self, broker_ids=None):
                 consumer_groups.update(response.groups)
         else:
             raise NotImplementedError(
-                "Support for ListGroups v{} has not yet been added to KafkaAdmin."
+                "Support for ListGroups v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         return list(consumer_groups)
 
@@ -744,7 +744,7 @@ def list_consumer_group_offsets(self, group_id, group_coordinator_id=None,
                     group_offsets_listing[TopicPartition(topic, partition)] = OffsetAndMetadata(offset, metadata)
         else:
             raise NotImplementedError(
-                "Support for OffsetFetch v{} has not yet been added to KafkaAdmin."
+                "Support for OffsetFetch v{} has not yet been added to KafkaAdminClient."
                 .format(version))
         return group_offsets_listing