* Fixed logger not working when provided as an argument to AdminClient

* Updated examples/adminapi.py to include usage of the custom logger with AdminClient

Author: pranavrth

CHANGELOG.md

@@ -7,6 +7,8 @@ v2.4.1 is a maintenance release with the following fixes and enhancements:
  - Removed usage of `strcpy` to enhance security of the client (#1745)
  - Fixed invalid write in `OAUTHBEARER/OIDC` extensions copy (#1745)
  - Fixed an issue related to import error of `TopicCollection` and `TopicPartitionInfo` classes when importing through other module like mypy.
+ - Fixed `logger` not working when provided as an argument to `AdminClient`
+ - Updated `examples/adminapi.py` to include usage of the custom logger with `AdminClient` 
 
 confluent-kafka-python is based on librdkafka v2.4.1, see the
 [librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.4.1)

examples/adminapi.py

@@ -890,8 +890,16 @@ def example_list_offsets(a, args):
     operation = sys.argv[2]
     args = sys.argv[3:]
 
+    # Custom logger
+    logger = logging.getLogger('AdminClient')
+    logger.setLevel(logging.INFO)
+    handler = logging.StreamHandler()
+    handler.setFormatter(logging.Formatter('%(asctime)-15s %(levelname)-8s %(message)s'))
+    logger.addHandler(handler)
+
     # Create Admin client
-    a = AdminClient({'bootstrap.servers': broker})
+    a = AdminClient({'bootstrap.servers': broker},
+                     logger=logger)
 
     opsmap = {'create_topics': example_create_topics,
               'delete_topics': example_delete_topics,
@@ -920,3 +928,6 @@ def example_list_offsets(a, args):
         sys.exit(1)
 
     opsmap[operation](a, args)
+
+    # Log messages through custom logger if provided
+    a.poll(0)

src/confluent_kafka/admin/__init__.py

@@ -109,16 +109,21 @@ class AdminClient (_AdminClientImpl):
     Requires broker version v0.11.0.0 or later.
     """
 
-    def __init__(self, conf):
+    def __init__(self, conf, logger=None):
         """
         Create a new AdminClient using the provided configuration dictionary.
 
         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
 
+        :param dict config: 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
         super(AdminClient, self).__init__(conf)
 
     @staticmethod

tests/test_log.py

@@ -3,6 +3,7 @@
 from io import StringIO
 import confluent_kafka
 import confluent_kafka.avro
+import confluent_kafka.admin
 import logging
 
 
@@ -158,3 +159,43 @@ def test_consumer_logger_logging_in_given_format():
     c.close()
 
     assert "Consumer Logger | INIT" in logMessage
+
+
+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)
+
+    admin_client = confluent_kafka.admin.AdminClient(
+        {"bootstrap.servers": "test", "logger": logger, "debug": "admin"})
+    admin_client.poll(0)
+
+    logMessage = stringBuffer.getvalue().strip()
+    stringBuffer.close()
+
+    assert "Admin Logger | INIT" in logMessage
+
+
+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)
+
+    admin_client = confluent_kafka.admin.AdminClient(
+        {"bootstrap.servers": "test", "debug": "admin"}, logger=logger)
+    admin_client.poll(0)
+
+    logMessage = stringBuffer.getvalue().strip()
+    stringBuffer.close()
+
+    assert "Admin Logger | INIT" in logMessage