PR comments

Author: pranavrth

examples/adminapi.py

@@ -899,7 +899,7 @@ def example_list_offsets(a, args):
 
     # Create Admin client
     a = AdminClient({'bootstrap.servers': broker},
-                     logger=logger)
+                    logger=logger)
 
     opsmap = {'create_topics': example_create_topics,
               'delete_topics': example_delete_topics,
@@ -930,4 +930,4 @@ def example_list_offsets(a, args):
     opsmap[operation](a, args)
 
     # Log messages through custom logger if provided
-    a.poll(0)
+    a.poll(0)

src/confluent_kafka/admin/__init__.py

@@ -115,12 +115,10 @@ def __init__(self, conf, logger=None):
 
         The AdminClient is a standard Kafka protocol client, supporting
         the standard librdkafka configuration properties as specified at
-        https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
+        https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md
 
-        :param dict config: Configuration properties. At a minimum ``bootstrap.servers`` **should** be set\n"
+        :param dict conf: Configuration properties. At a minimum ``bootstrap.servers`` **should** be set\n"
         :param Logger logger: Optional Logger instance to use as a custom log messages handler.
-
-        At least 'bootstrap.servers' should be configured.
         """
         if logger is not None:
             conf['logger'] = logger

tests/test_log.py

@@ -18,6 +18,16 @@ def filter(self, record):
         print(record)
 
 
+def _setup_string_buffer_logger():
+    stringBuffer = StringIO()
+    logger = logging.getLogger('Producer')
+    logger.setLevel(logging.DEBUG)
+    handler = logging.StreamHandler(stringBuffer)
+    handler.setFormatter(logging.Formatter('%(name)s Logger | %(message)s'))
+    logger.addHandler(handler)
+    return stringBuffer,logger
+
+
 def test_logging_consumer():
     """ Tests that logging works """
 
@@ -121,12 +131,7 @@ def test_logging_constructor():
 def test_producer_logger_logging_in_given_format():
     """Test that asserts that logging is working by matching part of the log message"""
 
-    stringBuffer = StringIO()
-    logger = logging.getLogger('Producer')
-    logger.setLevel(logging.DEBUG)
-    handler = logging.StreamHandler(stringBuffer)
-    handler.setFormatter(logging.Formatter('%(name)s Logger | %(message)s'))
-    logger.addHandler(handler)
+    stringBuffer, logger = _setup_string_buffer_logger()
 
     p = confluent_kafka.Producer(
         {"bootstrap.servers": "test", "logger": logger, "debug": "msg"})
@@ -143,12 +148,7 @@ def test_producer_logger_logging_in_given_format():
 def test_consumer_logger_logging_in_given_format():
     """Test that asserts that logging is working by matching part of the log message"""
 
-    stringBuffer = StringIO()
-    logger = logging.getLogger('Consumer')
-    logger.setLevel(logging.DEBUG)
-    handler = logging.StreamHandler(stringBuffer)
-    handler.setFormatter(logging.Formatter('%(name)s Logger | %(message)s'))
-    logger.addHandler(handler)
+    stringBuffer, logger = _setup_string_buffer_logger()
 
     c = confluent_kafka.Consumer(
         {"bootstrap.servers": "test", "group.id": "test", "logger": logger, "debug": "msg"})
@@ -164,12 +164,7 @@ def test_consumer_logger_logging_in_given_format():
 def test_admin_logger_logging_in_given_format_when_provided_in_conf():
     """Test that asserts that logging is working by matching part of the log message"""
 
-    stringBuffer = StringIO()
-    logger = logging.getLogger('Admin')
-    logger.setLevel(logging.DEBUG)
-    handler = logging.StreamHandler(stringBuffer)
-    handler.setFormatter(logging.Formatter('%(name)s Logger | %(message)s'))
-    logger.addHandler(handler)
+    stringBuffer, logger = _setup_string_buffer_logger()
 
     admin_client = confluent_kafka.admin.AdminClient(
         {"bootstrap.servers": "test", "logger": logger, "debug": "admin"})
@@ -184,12 +179,7 @@ def test_admin_logger_logging_in_given_format_when_provided_in_conf():
 def test_admin_logger_logging_in_given_format_when_provided_as_admin_client_argument():
     """Test that asserts that logging is working by matching part of the log message"""
 
-    stringBuffer = StringIO()
-    logger = logging.getLogger('Admin')
-    logger.setLevel(logging.DEBUG)
-    handler = logging.StreamHandler(stringBuffer)
-    handler.setFormatter(logging.Formatter('%(name)s Logger | %(message)s'))
-    logger.addHandler(handler)
+    stringBuffer, logger = _setup_string_buffer_logger()
 
     admin_client = confluent_kafka.admin.AdminClient(
         {"bootstrap.servers": "test", "debug": "admin"}, logger=logger)
@@ -198,4 +188,4 @@ def test_admin_logger_logging_in_given_format_when_provided_as_admin_client_argu
     logMessage = stringBuffer.getvalue().strip()
     stringBuffer.close()
 
-    assert "Admin Logger | INIT" in logMessage
+    assert "Admin Logger | INIT" in logMessage