Allow Kafka producer headers to be dict or list

CHANGELOG.md

@@ -61,6 +61,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1512](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1512))
 - Add commit method for ConfluentKafkaInstrumentor's ProxiedConsumer
   ([#1656](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1656))
+- Fix confluent-kafka instrumentation by allowing Producer headers to be dict or list
+  ([#1655](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1655))
 
 ## Version 1.15.0/0.36b0 (2022-12-10)
 

instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py

@@ -60,7 +60,12 @@ def set(self, carrier: textmap.CarrierT, key: str, value: str) -> None:
 
         if value:
             value = value.encode()
-        carrier.append((key, value))
+
+        if isinstance(carrier, list):
+            carrier.append((key, value))
+
+        if isinstance(carrier, dict):
+            carrier[key] = value
 
 
 _kafka_getter = KafkaContextGetter()

instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py

@@ -23,6 +23,9 @@
     ProxiedConsumer,
     ProxiedProducer,
 )
+from opentelemetry.instrumentation.confluent_kafka.utils import (
+    KafkaContextSetter,
+)
 
 
 class TestConfluentKafka(TestCase):
@@ -73,3 +76,16 @@ def test_consumer_commit_method_exists(self) -> None:
         consumer = instrumentation.instrument_consumer(consumer)
         self.assertEqual(consumer.__class__, ProxiedConsumer)
         self.assertTrue(hasattr(consumer, "commit"))
+
+    def test_context_setter(self) -> None:
+        context_setter = KafkaContextSetter()
+
+        carrier_dict = {"key1": "val1"}
+        context_setter.set(carrier_dict, "key2", "val2")
+        self.assertGreaterEqual(
+            carrier_dict.items(), {"key2": "val2".encode()}.items()
+        )
+
+        carrier_list = [("key1", "val1")]
+        context_setter.set(carrier_list, "key2", "val2")
+        self.assertTrue(("key2", "val2".encode()) in carrier_list)