fix(kafka): allow producer/consumer initialized with unpacked config [APMS-15618] [backport 2.21] (#13316)

Backport 02da078 from #13303 to 2.21.

Fixes an issue where Kafka connection initialization failed when passing
in an unpacked config into Producer or Consumer. confluent-kafka
supports both unpacked config and dict config.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Quinna Halim <[email protected]>

Author: github-actions[bot]

ddtrace/contrib/internal/kafka/patch.py

@@ -62,7 +62,9 @@ def get_version():
 
 
 class TracedProducerMixin:
-    def __init__(self, config, *args, **kwargs):
+    def __init__(self, config=None, *args, **kwargs):
+        if not config:
+            config = kwargs
         super(TracedProducerMixin, self).__init__(config, *args, **kwargs)
         self._dd_bootstrap_servers = (
             config.get("bootstrap.servers")
@@ -79,7 +81,9 @@ def __bool__(self):
 
 
 class TracedConsumerMixin:
-    def __init__(self, config, *args, **kwargs):
+    def __init__(self, config=None, *args, **kwargs):
+        if not config:
+            config = kwargs
         super(TracedConsumerMixin, self).__init__(config, *args, **kwargs)
         self._group_id = config.get("group.id", "")
         self._auto_commit = asbool(config.get("enable.auto.commit", True))

releasenotes/notes/fix-consumer-producer-config-initialize-6f17da8b3a806b7e.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    kafka: Fixes an issue where a producer or consumer initialized with an unpacked config resulted in TypeError, causing a failed connection.
+    confluent-kafka supports both unpacked and packed config; this change allows initialization with either.

tests/contrib/kafka/test_kafka.py

@@ -12,6 +12,7 @@
 import pytest
 
 from ddtrace.contrib.internal.kafka.patch import TracedConsumer
+from ddtrace.contrib.internal.kafka.patch import TracedProducer
 from ddtrace.contrib.internal.kafka.patch import patch
 from ddtrace.contrib.internal.kafka.patch import unpatch
 import ddtrace.internal.datastreams  # noqa: F401 - used as part of mock patching
@@ -211,6 +212,19 @@ def test_consumer_created_with_logger_does_not_raise(tracer):
     consumer.close()
 
 
+def test_consumer_initialized_with_unpacked_config(tracer):
+    """Test that adding a logger to a Consumer init does not raise any errors."""
+    consumer = confluent_kafka.Consumer(
+        **{
+            "bootstrap.servers": BOOTSTRAP_SERVERS,
+            "group.id": GROUP_ID,
+            "auto.offset.reset": "earliest",
+        },
+    )
+    assert isinstance(consumer, TracedConsumer)
+    consumer.close()
+
+
 def test_empty_list_from_consume_does_not_raise():
     # https://github.com/DataDog/dd-trace-py/issues/8846
     patch()
@@ -247,6 +261,23 @@ def test_producer_bootstrap_servers(config, expect_servers, tracer):
         assert producer._dd_bootstrap_servers is None
 
 
+@pytest.mark.parametrize(
+    "config,expect_servers",
+    [
+        ({"bootstrap.servers": BOOTSTRAP_SERVERS}, BOOTSTRAP_SERVERS),
+        ({"metadata.broker.list": BOOTSTRAP_SERVERS}, BOOTSTRAP_SERVERS),
+        ({}, None),
+    ],
+)
+def test_producer_initialized_unpacked_config(config, expect_servers, tracer):
+    producer = confluent_kafka.Producer(**config)
+    assert isinstance(producer, TracedProducer)
+    if expect_servers is not None:
+        assert producer._dd_bootstrap_servers == expect_servers
+    else:
+        assert producer._dd_bootstrap_servers is None
+
+
 def test_produce_single_server(dummy_tracer, producer, kafka_topic):
     Pin.override(producer, tracer=dummy_tracer)
     producer.produce(kafka_topic, PAYLOAD, key=KEY)