fix(kafka): pass *args, **kwargs to TracedConsumer.__init__ (#5887)

Pass arbitrary positional and keyword arguments to `TracedConsumer` so
as to allow client to configure their instance of `KafkaConsumer`. This
is, for example, [required for configuring a
logger](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kafka-client-configuration)
internal to the Consumer, where a logger can only be passed as a kwarg.

Currently, passing such a kwarg leads to the following error:
```TypeError: TracedConsumer.__init__() got an unexpected keyword argument 'logger'```

Related to issue [#5837](#5873)

---

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included in the PR.
- [x] Risk is outlined (performance impact, potential for breakage, maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note guidelines](https://ddtrace.readthedocs.io/en/stable/contributing.html#Release-Note-Guidelines) are followed.
- [x] Documentation is included (in-code, generated user docs, [public corp docs](https://github.com/DataDog/documentation/)).
- [x] OPTIONAL: PR description includes explicit acknowledgement of the performance implications of the change as reported in the benchmarks PR comment.

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking [API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces) changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance implications of this PR as reported in the benchmarks PR comment.

---------

Co-authored-by: Yun Kim <[email protected]>
Co-authored-by: Munir Abdinur <[email protected]>
Co-authored-by: Munir Abdinur <[email protected]>

Author: 3 people

ddtrace/contrib/kafka/patch.py

@@ -39,8 +39,8 @@ def __bool__(self):
 
 
 class TracedConsumer(confluent_kafka.Consumer):
-    def __init__(self, config):
-        super(TracedConsumer, self).__init__(config)
+    def __init__(self, config, *args, **kwargs):
+        super(TracedConsumer, self).__init__(config, *args, **kwargs)
         self._group_id = config["group.id"]
 
     def poll(self, timeout=1):

releasenotes/notes/pass-args-kwargs-to-kafka-f3ca8128a2c4d612.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    kafka: Fixes ``TypeError` raised when artibrary keyword arguments are passed to ``confluent_kafka.Consumer``

tests/contrib/kafka/test_kafka.py

@@ -1,3 +1,5 @@
+import logging
+
 import confluent_kafka
 import pytest
 import six
@@ -63,6 +65,21 @@ def consumer(tracer):
     _consumer.close()
 
 
+def test_consumer_created_with_logger_does_not_raise(tracer):
+    """Test that adding a logger to a Consumer init does not raise any errors."""
+    logger = logging.getLogger()
+    # regression test for DataDog/dd-trace-py/issues/5873
+    consumer = confluent_kafka.Consumer(
+        {
+            "bootstrap.servers": BOOTSTRAP_SERVERS,
+            "group.id": GROUP_ID,
+            "auto.offset.reset": "earliest",
+        },
+        logger=logger,
+    )
+    consumer.close()
+
+
 @pytest.mark.parametrize("tombstone", [False, True])
 @pytest.mark.snapshot(ignores=["metrics.kafka.message_offset"])
 def test_message(producer, consumer, tombstone):