Moved defaults to the class attribute.

Author: Julian

kafka/consumer/group.py

@@ -166,6 +166,7 @@ class KafkaConsumer(six.Iterator):
     """
     DEFAULT_CONFIG = {
         'bootstrap_servers': 'localhost',
+        'client': None,
         'client_id': 'kafka-python-' + __version__,
         'group_id': 'kafka-python-default-group',
         'key_deserializer': None,
@@ -204,8 +205,6 @@ class KafkaConsumer(six.Iterator):
     }
 
     def __init__(self, *topics, **configs):
-        client = configs.pop("client", None)
-
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
             if key in configs:
@@ -231,9 +230,7 @@ def __init__(self, *topics, **configs):
         metric_group_prefix = 'consumer'
         # TODO _metrics likely needs to be passed to KafkaClient, etc.
 
-        if client is None:
-            client = KafkaClient(**self.config)
-        self._client = client
+        self._client = self.config['client'] or KafkaClient(**self.config)
 
         # Check Broker Version if not set explicitly
         if self.config['api_version'] == 'auto':

kafka/producer/kafka.py

@@ -226,6 +226,7 @@ class KafkaProducer(object):
     """
     _DEFAULT_CONFIG = {
         'bootstrap_servers': 'localhost',
+        'client': None,
         'client_id': None,
         'key_serializer': None,
         'value_serializer': None,
@@ -256,7 +257,7 @@ class KafkaProducer(object):
         'api_version': 'auto',
     }
 
-    def __init__(self, client=None, **configs):
+    def __init__(self, **configs):
         log.debug("Starting the Kafka producer") # trace
         self.config = copy.copy(self._DEFAULT_CONFIG)
         for key in self.config:
@@ -273,9 +274,7 @@ def __init__(self, client=None, **configs):
         if self.config['acks'] == 'all':
             self.config['acks'] = -1
 
-        if client is None:
-            client = KafkaClient(**self.config)
-        self.client = client
+        self.client = self.config['client'] or KafkaClient(**self.config)
 
         # Check Broker Version if not set explicitly
         if self.config['api_version'] == 'auto':