From e560815d557e491df818ec51293a962aceb04ba8 Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Sun, 18 Oct 2026 03:39:19 +0530
Subject: [PATCH 01/10] feat(instrumentation-confluent-kafka): patch
 AIOProducer and AIOConsumer

---
 .../confluent_kafka/__init__.py               | 131 ++++++++++++++++++
 .../tests/test_instrumentation.py             |  58 ++++++++
 .../tests/utils.py                            |  34 +++++
 3 files changed, 223 insertions(+)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 3637f9b6386..7549ec7377c 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -102,6 +102,17 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ...
 import confluent_kafka
 import wrapt
 from confluent_kafka import Consumer, Producer
+try:
+    from confluent_kafka.aio import AIOConsumer, AIOProducer
+except ImportError:
+    try:
+        from confluent_kafka.experimental.aio import (
+            AIOConsumer,
+            AIOProducer,
+        )
+    except ImportError:
+        AIOProducer = None
+        AIOConsumer = None
 
 from opentelemetry import context, propagate, trace
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
@@ -154,6 +165,7 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.config = _capture_config(args, kwargs)
         self._current_consume_span = None
+        
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation
@@ -167,6 +179,36 @@ def consume(self, *args, **kwargs):  # pylint: disable=useless-super-delegation
     def close(self):  # pylint: disable=useless-super-delegation
         return super().close()
 
+# Instrument asynchronous AIOProducer and AIOConsumer if available in confluent-kafka
+if AIOProducer is not None:
+    class AutoInstrumentedAIOProducer(AIOProducer):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.config = _capture_config(args, kwargs)
+
+        # Deliberately implemented so wrapt can intercept the async produce call
+        async def produce(self, topic, value=None, *args, **kwargs):
+            return await super().produce(topic, value, *args, **kwargs)
+else:
+    AutoInstrumentedAIOProducer = None
+
+
+if AIOConsumer is not None:
+    class AutoInstrumentedAIOConsumer(AIOConsumer):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.config = _capture_config(args, kwargs)
+            self._current_consume_span = None
+
+        # Deliberately implemented so wrapt can intercept the async poll call
+        async def poll(self, timeout=-1):
+            return await super().poll(timeout)
+
+        # Deliberately implemented so wrapt can intercept the async close call
+        async def close(self):
+            return await super().close()
+else:
+    AutoInstrumentedAIOConsumer = None
 
 class ProxiedProducer(Producer):
     def __init__(self, producer: Producer, tracer: Tracer):
@@ -341,13 +383,44 @@ def _inner_wrap_close(func, instance, args, kwargs):
             "close",
             _inner_wrap_close,
         )
+        if AutoInstrumentedAIOProducer is not None:
+            def _inner_wrap_aio_produce(func, instance, args, kwargs):
+                return ConfluentKafkaInstrumentor.wrap_aio_produce(func, instance, self._tracer, args, kwargs)
+
+            wrapt.wrap_function_wrapper(
+                AutoInstrumentedAIOProducer,
+                "produce",
+                _inner_wrap_aio_produce,
+            )
+
+        if AutoInstrumentedAIOConsumer is not None:
+            def _inner_wrap_aio_poll(func, instance, args, kwargs):
+                return ConfluentKafkaInstrumentor.wrap_aio_poll(func, instance, self._tracer, args, kwargs)
+
+            def _inner_wrap_aio_close(func, instance, args, kwargs):
+                return ConfluentKafkaInstrumentor.wrap_aio_close(func, instance, args, kwargs)
 
+            wrapt.wrap_function_wrapper(
+                AutoInstrumentedAIOConsumer,
+                "poll",
+                _inner_wrap_aio_poll,
+            )
+            wrapt.wrap_function_wrapper(
+                AutoInstrumentedAIOConsumer,
+                "close",
+                _inner_wrap_aio_close,
+            )         
     def _uninstrument(self, **kwargs):
         confluent_kafka.Producer = self._original_kafka_producer
         confluent_kafka.Consumer = self._original_kafka_consumer
 
         unwrap(AutoInstrumentedProducer, "produce")
         unwrap(AutoInstrumentedConsumer, "poll")
+        if AutoInstrumentedAIOProducer is not None:
+            unwrap(AutoInstrumentedAIOProducer, "produce")
+        if AutoInstrumentedAIOConsumer is not None:
+            unwrap(AutoInstrumentedAIOConsumer, "poll")
+            unwrap(AutoInstrumentedAIOConsumer, "close")
 
     @staticmethod
     def wrap_produce(func, instance, tracer, args, kwargs):
@@ -420,6 +493,64 @@ def wrap_consume(func, instance, tracer, args, kwargs):
 
     @staticmethod
     def wrap_close(func, instance, args, kwargs):
+
+        
         if instance._current_consume_span:
             _end_current_consume_span(instance)
         func(*args, **kwargs)
+
+
+    @staticmethod
+    async def wrap_aio_produce(func, instance, tracer, args, kwargs):
+        topic = kwargs.get("topic")
+        if not topic:
+            topic = args[0]
+
+        span_name = _get_span_name("send", topic)
+        with tracer.start_as_current_span(name=span_name, kind=trace.SpanKind.PRODUCER) as span:
+            headers = KafkaPropertiesExtractor.extract_produce_headers(args, kwargs)
+            if headers is None:
+                headers = []
+                kwargs["headers"] = headers
+
+            topic = KafkaPropertiesExtractor.extract_produce_topic(args, kwargs)
+            bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
+            _enrich_span(
+                span,
+                topic,
+                operation=MessagingOperationTypeValues.PUBLISH,
+                bootstrap_servers=bootstrap_servers,
+            )
+            propagate.inject(
+                headers,
+                setter=_kafka_setter,
+            )
+            return await func(*args, **kwargs)
+
+    @staticmethod
+    async def wrap_aio_poll(func, instance, tracer, args, kwargs):
+        if instance._current_consume_span:
+            _end_current_consume_span(instance)
+
+        record = await func(*args, **kwargs)
+        if record:
+            bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
+            with tracer.start_as_current_span("recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER):
+                _create_new_consume_span(instance, tracer, [record])
+                _enrich_span(
+                    instance._current_consume_span,
+                    record.topic(),
+                    record.partition(),
+                    record.offset(),
+                    operation=MessagingOperationTypeValues.PROCESS,
+                    bootstrap_servers=bootstrap_servers,
+                )
+            instance._current_context_token = context.attach(trace.set_span_in_context(instance._current_consume_span))
+
+        return record
+
+    @staticmethod
+    async def wrap_aio_close(func, instance, args, kwargs):
+        if instance._current_consume_span:
+            _end_current_consume_span(instance)
+        return await func(*args, **kwargs)
\ No newline at end of file
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
index 78fbead5e7a..58de2fa3a14 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
@@ -470,3 +470,61 @@ def test_consumer_sets_bootstrap_servers_attributes(self) -> None:
         process_span = next(s for s in self.memory_exporter.get_finished_spans() if s.name == "topic-1 process")
         self.assertEqual(process_span.attributes[SERVER_ADDRESS], "broker-1")
         self.assertEqual(process_span.attributes[SERVER_PORT], 9092)
+
+    def test_aio_produce(self) -> None:
+        import asyncio
+        from .utils import MockedAIOProducer
+
+        async def _run():
+            instrumentation = ConfluentKafkaInstrumentor()
+            tracer = self.tracer_provider.get_tracer(__name__)
+            producer = MockedAIOProducer({"bootstrap.servers": "localhost:9092"})
+
+            await instrumentation.wrap_aio_produce(
+                producer.produce,
+                producer,
+                tracer,
+                args=(),
+                kwargs={"topic": "test-topic", "value": b"test-value"},
+            )
+
+            spans = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(spans), 1)
+            self.assertEqual(spans[0].name, "test-topic send")
+            self.assertEqual(len(producer._queue), 1)
+            headers = producer._queue[0].headers()
+            self.assertTrue(any("traceparent" in str(h) for h in headers))
+
+        asyncio.run(_run())
+
+    def test_aio_poll(self) -> None:
+        import asyncio
+        from .utils import MockedAIOConsumer, MockedMessage
+
+        async def _run():
+            instrumentation = ConfluentKafkaInstrumentor()
+            tracer = self.tracer_provider.get_tracer(__name__)
+            consumer = MockedAIOConsumer({"bootstrap.servers": "localhost:9092"})
+            consumer._queue.append(MockedMessage("test-topic", 0, 0, []))
+
+            msg = await instrumentation.wrap_aio_poll(
+                consumer.poll,
+                consumer,
+                tracer,
+                args=(),
+                kwargs={},
+            )
+            self.assertIsNotNone(msg)
+
+            await instrumentation.wrap_aio_poll(
+                consumer.poll,
+                consumer,
+                tracer,
+                args=(),
+                kwargs={},
+            )
+
+            spans = self.memory_exporter.get_finished_spans()
+            self.assertTrue(any(s.name == "test-topic process" for s in spans))
+
+        asyncio.run(_run())
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
index 86c360eeb9f..3868686ae02 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
@@ -81,3 +81,37 @@ def poll(self, *args, **kwargs):
 
     def flush(self, *args, **kwargs):
         return len(self._queue)
+
+
+
+class MockedAIOProducer:
+    def __init__(self, config=None):
+        self._queue = []
+        self.config = config or {}
+
+    async def produce(self, topic, value=None, *args, **kwargs):
+        msg = MockedMessage(
+            topic=topic,
+            partition=0,
+            offset=0,
+            headers=kwargs.get("headers", []),
+            key=kwargs.get("key"),
+            value=value,
+        )
+        self._queue.append(msg)
+        return msg
+
+
+class MockedAIOConsumer:
+    def __init__(self, config=None):
+        self._queue = []
+        self.config = config or {}
+        self._current_consume_span = None
+
+    async def poll(self, timeout=None):
+        if len(self._queue) > 0:
+            return self._queue.pop(0)
+        return None
+
+    async def close(self):
+        pass

From af2ec856dde976b9d7f64efa355f09e8c7ae35de Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Sun, 18 Oct 2026 04:04:57 +0530
Subject: [PATCH 02/10] docs: add changelog fragment for #5115

---
 .changelog/5115.added | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 .changelog/5115.added

diff --git a/.changelog/5115.added b/.changelog/5115.added
new file mode 100644
index 00000000000..a9b9de1c3a7
--- /dev/null
+++ b/.changelog/5115.added
@@ -0,0 +1 @@
+`opentelemetry-instrumentation-confluent-kafka`: add support for `AIOProducer` and `AIOConsumer`

From d16869c089eb46f66a12f4b70d6c4b57c017fc71 Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Thu, 29 Oct 2026 00:13:44 +0530
Subject: [PATCH 03/10] fix: resolve CI lint, format, and docs failures

- Remove trailing whitespace in __init__.py (ruff W291)
- Fix blank line counts: E302 (2 before top-level class) and
  E303 (too many blank lines inside class body)
- Add blank line before _uninstrument after wrapt block (E301)
- Remove extra blank line before MockedAIOProducer in tests/utils.py
- Fix changelog fragment line-ending (LF, not CRLF)
- Add AutoInstrumentedAIOProducer/AIOConsumer to docs/nitpick-exceptions.ini
  so Sphinx --nitpicky does not fail on unresolved class references
---
 docs/nitpick-exceptions.ini                               | 2 ++
 .../instrumentation/confluent_kafka/__init__.py           | 8 +++++---
 .../tests/utils.py                                        | 1 -
 3 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/docs/nitpick-exceptions.ini b/docs/nitpick-exceptions.ini
index ac6e29690af..1d3a0d5dbad 100644
--- a/docs/nitpick-exceptions.ini
+++ b/docs/nitpick-exceptions.ini
@@ -13,6 +13,8 @@ py-class=
     opentelemetry.sdk.trace.id_generator.IdGenerator
     opentelemetry.instrumentation.confluent_kafka.ProxiedProducer
     opentelemetry.instrumentation.confluent_kafka.ProxiedConsumer
+    opentelemetry.instrumentation.confluent_kafka.AutoInstrumentedAIOProducer
+    opentelemetry.instrumentation.confluent_kafka.AutoInstrumentedAIOConsumer
     opentelemetry.instrumentation.instrumentor.BaseInstrumentor
     ; - AwsXRayIdGenerator
     TextMapPropagator
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 7549ec7377c..8c6033692e2 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -165,7 +165,6 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.config = _capture_config(args, kwargs)
         self._current_consume_span = None
-        
 
     # This method is deliberately implemented in order to allow wrapt to wrap this function
     def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation
@@ -179,6 +178,7 @@ def consume(self, *args, **kwargs):  # pylint: disable=useless-super-delegation
     def close(self):  # pylint: disable=useless-super-delegation
         return super().close()
 
+
 # Instrument asynchronous AIOProducer and AIOConsumer if available in confluent-kafka
 if AIOProducer is not None:
     class AutoInstrumentedAIOProducer(AIOProducer):
@@ -210,6 +210,7 @@ async def close(self):
 else:
     AutoInstrumentedAIOConsumer = None
 
+
 class ProxiedProducer(Producer):
     def __init__(self, producer: Producer, tracer: Tracer):
         self._producer = producer
@@ -409,7 +410,8 @@ def _inner_wrap_aio_close(func, instance, args, kwargs):
                 AutoInstrumentedAIOConsumer,
                 "close",
                 _inner_wrap_aio_close,
-            )         
+            )
+
     def _uninstrument(self, **kwargs):
         confluent_kafka.Producer = self._original_kafka_producer
         confluent_kafka.Consumer = self._original_kafka_consumer
@@ -494,7 +496,7 @@ def wrap_consume(func, instance, tracer, args, kwargs):
     @staticmethod
     def wrap_close(func, instance, args, kwargs):
 
-        
+
         if instance._current_consume_span:
             _end_current_consume_span(instance)
         func(*args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
index 3868686ae02..b2fd638d59a 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
@@ -83,7 +83,6 @@ def flush(self, *args, **kwargs):
         return len(self._queue)
 
 
-
 class MockedAIOProducer:
     def __init__(self, config=None):
         self._queue = []

From 0e7165deb149fe02a5f6b88fce24c7c3d5bab0ad Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Thu, 29 Oct 2026 14:57:36 +0530
Subject: [PATCH 04/10] fix(confluent-kafka): patch public AIO classes and
 avoid forcing headers

- Replace public AIOProducer and AIOConsumer on module and restore on uninstrument
- Do not force kwargs['headers'] = [] in wrap_aio_produce if headers is None, avoiding NotImplementedError
- Add test_aio_instrument_and_uninstrument and update test_aio_produce
---
 .../confluent_kafka/__init__.py               | 50 ++++++++++++----
 .../tests/test_instrumentation.py             | 60 +++++++++++++++----
 2 files changed, 89 insertions(+), 21 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 8c6033692e2..9d22b4b4ede 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -102,10 +102,13 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ...
 import confluent_kafka
 import wrapt
 from confluent_kafka import Consumer, Producer
+
 try:
+    import confluent_kafka.aio as _aio_module
     from confluent_kafka.aio import AIOConsumer, AIOProducer
 except ImportError:
     try:
+        import confluent_kafka.experimental.aio as _aio_module
         from confluent_kafka.experimental.aio import (
             AIOConsumer,
             AIOProducer,
@@ -113,6 +116,7 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ...
     except ImportError:
         AIOProducer = None
         AIOConsumer = None
+        _aio_module = None
 
 from opentelemetry import context, propagate, trace
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
@@ -181,6 +185,7 @@ def close(self):  # pylint: disable=useless-super-delegation
 
 # Instrument asynchronous AIOProducer and AIOConsumer if available in confluent-kafka
 if AIOProducer is not None:
+
     class AutoInstrumentedAIOProducer(AIOProducer):
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
@@ -194,6 +199,7 @@ async def produce(self, topic, value=None, *args, **kwargs):
 
 
 if AIOConsumer is not None:
+
     class AutoInstrumentedAIOConsumer(AIOConsumer):
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
@@ -339,6 +345,21 @@ def _instrument(self, **kwargs):
         confluent_kafka.Producer = AutoInstrumentedProducer
         confluent_kafka.Consumer = AutoInstrumentedConsumer
 
+        if _aio_module is not None:
+            if hasattr(_aio_module, "AIOProducer") and AutoInstrumentedAIOProducer is not None:
+                self._original_kafka_aio_producer = _aio_module.AIOProducer
+                _aio_module.AIOProducer = AutoInstrumentedAIOProducer
+            if hasattr(_aio_module, "AIOConsumer") and AutoInstrumentedAIOConsumer is not None:
+                self._original_kafka_aio_consumer = _aio_module.AIOConsumer
+                _aio_module.AIOConsumer = AutoInstrumentedAIOConsumer
+
+        if hasattr(confluent_kafka, "AIOProducer") and AutoInstrumentedAIOProducer is not None:
+            self._original_kafka_top_aio_producer = confluent_kafka.AIOProducer
+            confluent_kafka.AIOProducer = AutoInstrumentedAIOProducer
+        if hasattr(confluent_kafka, "AIOConsumer") and AutoInstrumentedAIOConsumer is not None:
+            self._original_kafka_top_aio_consumer = confluent_kafka.AIOConsumer
+            confluent_kafka.AIOConsumer = AutoInstrumentedAIOConsumer
+
         tracer_provider = kwargs.get("tracer_provider")
         tracer = trace.get_tracer(
             __name__,
@@ -385,6 +406,7 @@ def _inner_wrap_close(func, instance, args, kwargs):
             _inner_wrap_close,
         )
         if AutoInstrumentedAIOProducer is not None:
+
             def _inner_wrap_aio_produce(func, instance, args, kwargs):
                 return ConfluentKafkaInstrumentor.wrap_aio_produce(func, instance, self._tracer, args, kwargs)
 
@@ -395,6 +417,7 @@ def _inner_wrap_aio_produce(func, instance, args, kwargs):
             )
 
         if AutoInstrumentedAIOConsumer is not None:
+
             def _inner_wrap_aio_poll(func, instance, args, kwargs):
                 return ConfluentKafkaInstrumentor.wrap_aio_poll(func, instance, self._tracer, args, kwargs)
 
@@ -416,6 +439,17 @@ def _uninstrument(self, **kwargs):
         confluent_kafka.Producer = self._original_kafka_producer
         confluent_kafka.Consumer = self._original_kafka_consumer
 
+        if _aio_module is not None:
+            if hasattr(self, "_original_kafka_aio_producer") and self._original_kafka_aio_producer is not None:
+                _aio_module.AIOProducer = self._original_kafka_aio_producer
+            if hasattr(self, "_original_kafka_aio_consumer") and self._original_kafka_aio_consumer is not None:
+                _aio_module.AIOConsumer = self._original_kafka_aio_consumer
+
+        if hasattr(self, "_original_kafka_top_aio_producer") and self._original_kafka_top_aio_producer is not None:
+            confluent_kafka.AIOProducer = self._original_kafka_top_aio_producer
+        if hasattr(self, "_original_kafka_top_aio_consumer") and self._original_kafka_top_aio_consumer is not None:
+            confluent_kafka.AIOConsumer = self._original_kafka_top_aio_consumer
+
         unwrap(AutoInstrumentedProducer, "produce")
         unwrap(AutoInstrumentedConsumer, "poll")
         if AutoInstrumentedAIOProducer is not None:
@@ -496,12 +530,10 @@ def wrap_consume(func, instance, tracer, args, kwargs):
     @staticmethod
     def wrap_close(func, instance, args, kwargs):
 
-
         if instance._current_consume_span:
             _end_current_consume_span(instance)
         func(*args, **kwargs)
 
-
     @staticmethod
     async def wrap_aio_produce(func, instance, tracer, args, kwargs):
         topic = kwargs.get("topic")
@@ -511,9 +543,11 @@ async def wrap_aio_produce(func, instance, tracer, args, kwargs):
         span_name = _get_span_name("send", topic)
         with tracer.start_as_current_span(name=span_name, kind=trace.SpanKind.PRODUCER) as span:
             headers = KafkaPropertiesExtractor.extract_produce_headers(args, kwargs)
-            if headers is None:
-                headers = []
-                kwargs["headers"] = headers
+            if headers is not None:
+                propagate.inject(
+                    headers,
+                    setter=_kafka_setter,
+                )
 
             topic = KafkaPropertiesExtractor.extract_produce_topic(args, kwargs)
             bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
@@ -523,10 +557,6 @@ async def wrap_aio_produce(func, instance, tracer, args, kwargs):
                 operation=MessagingOperationTypeValues.PUBLISH,
                 bootstrap_servers=bootstrap_servers,
             )
-            propagate.inject(
-                headers,
-                setter=_kafka_setter,
-            )
             return await func(*args, **kwargs)
 
     @staticmethod
@@ -555,4 +585,4 @@ async def wrap_aio_poll(func, instance, tracer, args, kwargs):
     async def wrap_aio_close(func, instance, args, kwargs):
         if instance._current_consume_span:
             _end_current_consume_span(instance)
-        return await func(*args, **kwargs)
\ No newline at end of file
+        return await func(*args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
index 58de2fa3a14..84ae00219b1 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
@@ -1,8 +1,8 @@
+import asyncio
+
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
-
 # pylint: disable=no-name-in-module,import-outside-toplevel
-
 from opentelemetry.instrumentation.confluent_kafka import (
     AutoInstrumentedConsumer,
     AutoInstrumentedProducer,
@@ -472,34 +472,44 @@ def test_consumer_sets_bootstrap_servers_attributes(self) -> None:
         self.assertEqual(process_span.attributes[SERVER_PORT], 9092)
 
     def test_aio_produce(self) -> None:
-        import asyncio
-        from .utils import MockedAIOProducer
+        from .utils import MockedAIOProducer  # noqa: PLC0415
 
         async def _run():
             instrumentation = ConfluentKafkaInstrumentor()
             tracer = self.tracer_provider.get_tracer(__name__)
             producer = MockedAIOProducer({"bootstrap.servers": "localhost:9092"})
 
+            # Without headers: verify produce succeeds and headers kwarg is NOT forced
+            # (confluent-kafka AIOProducer raises NotImplementedError if headers is passed)
+            call_kwargs = {"topic": "test-topic", "value": b"test-value"}
             await instrumentation.wrap_aio_produce(
                 producer.produce,
                 producer,
                 tracer,
                 args=(),
-                kwargs={"topic": "test-topic", "value": b"test-value"},
+                kwargs=call_kwargs,
             )
-
+            self.assertNotIn("headers", call_kwargs)
             spans = self.memory_exporter.get_finished_spans()
             self.assertEqual(len(spans), 1)
             self.assertEqual(spans[0].name, "test-topic send")
-            self.assertEqual(len(producer._queue), 1)
-            headers = producer._queue[0].headers()
-            self.assertTrue(any("traceparent" in str(h) for h in headers))
+
+            # With headers: verify traceparent is injected into supplied headers
+            self.memory_exporter.clear()
+            supplied_headers = []
+            await instrumentation.wrap_aio_produce(
+                producer.produce,
+                producer,
+                tracer,
+                args=(),
+                kwargs={"topic": "test-topic", "value": b"test-value", "headers": supplied_headers},
+            )
+            self.assertTrue(any("traceparent" in str(h) for h in supplied_headers))
 
         asyncio.run(_run())
 
     def test_aio_poll(self) -> None:
-        import asyncio
-        from .utils import MockedAIOConsumer, MockedMessage
+        from .utils import MockedAIOConsumer, MockedMessage  # noqa: PLC0415
 
         async def _run():
             instrumentation = ConfluentKafkaInstrumentor()
@@ -528,3 +538,31 @@ async def _run():
             self.assertTrue(any(s.name == "test-topic process" for s in spans))
 
         asyncio.run(_run())
+
+    def test_aio_instrument_and_uninstrument(self) -> None:
+        from opentelemetry.instrumentation.confluent_kafka import (  # noqa: PLC0415
+            AutoInstrumentedAIOConsumer,
+            AutoInstrumentedAIOProducer,
+            _aio_module,
+        )
+
+        if _aio_module is None or AutoInstrumentedAIOProducer is None:
+            return
+
+        instrumentor = ConfluentKafkaInstrumentor()
+        orig_producer = getattr(_aio_module, "AIOProducer", None)
+        orig_consumer = getattr(_aio_module, "AIOConsumer", None)
+
+        instrumentor.instrument()
+        try:
+            if orig_producer is not None:
+                self.assertIs(_aio_module.AIOProducer, AutoInstrumentedAIOProducer)
+            if orig_consumer is not None:
+                self.assertIs(_aio_module.AIOConsumer, AutoInstrumentedAIOConsumer)
+        finally:
+            instrumentor.uninstrument()
+
+        if orig_producer is not None:
+            self.assertIs(_aio_module.AIOProducer, orig_producer)
+        if orig_consumer is not None:
+            self.assertIs(_aio_module.AIOConsumer, orig_consumer)

From a08823cc0e4f65f8ec234475ea693c6d24edfe04 Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Sat, 31 Oct 2026 01:25:07 +0530
Subject: [PATCH 05/10] fix: avoid duplicate AIO consumer spans, fix license
 header, pylint W1113 and CRLF line endings

---
 .../confluent_kafka/__init__.py               | 28 ++++------------
 .../tests/test_instrumentation.py             | 32 ++++++++++++++++---
 .../tests/utils.py                            | 22 +++++++++----
 3 files changed, 48 insertions(+), 34 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 9d22b4b4ede..9d28edd2a42 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -204,7 +204,8 @@ class AutoInstrumentedAIOConsumer(AIOConsumer):
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
             self.config = _capture_config(args, kwargs)
-            self._current_consume_span = None
+            # Span lifecycle is managed by the inner Consumer layer, which is
+            # already instrumented via AutoInstrumentedConsumer.
 
         # Deliberately implemented so wrapt can intercept the async poll call
         async def poll(self, timeout=-1):
@@ -561,28 +562,11 @@ async def wrap_aio_produce(func, instance, tracer, args, kwargs):
 
     @staticmethod
     async def wrap_aio_poll(func, instance, tracer, args, kwargs):
-        if instance._current_consume_span:
-            _end_current_consume_span(instance)
-
-        record = await func(*args, **kwargs)
-        if record:
-            bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(instance)
-            with tracer.start_as_current_span("recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER):
-                _create_new_consume_span(instance, tracer, [record])
-                _enrich_span(
-                    instance._current_consume_span,
-                    record.topic(),
-                    record.partition(),
-                    record.offset(),
-                    operation=MessagingOperationTypeValues.PROCESS,
-                    bootstrap_servers=bootstrap_servers,
-                )
-            instance._current_context_token = context.attach(trace.set_span_in_context(instance._current_consume_span))
-
-        return record
+        # AIOConsumer delegates to an inner Consumer that is already instrumented
+        # via AutoInstrumentedConsumer. Creating spans here would produce duplicates,
+        # so this wrapper is intentionally a pass-through.
+        return await func(*args, **kwargs)
 
     @staticmethod
     async def wrap_aio_close(func, instance, args, kwargs):
-        if instance._current_consume_span:
-            _end_current_consume_span(instance)
         return await func(*args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
index 84ae00219b1..fdd733a49f5 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
@@ -1,8 +1,8 @@
-import asyncio
-
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 # pylint: disable=no-name-in-module,import-outside-toplevel
+import asyncio
+
 from opentelemetry.instrumentation.confluent_kafka import (
     AutoInstrumentedConsumer,
     AutoInstrumentedProducer,
@@ -514,9 +514,28 @@ def test_aio_poll(self) -> None:
         async def _run():
             instrumentation = ConfluentKafkaInstrumentor()
             tracer = self.tracer_provider.get_tracer(__name__)
-            consumer = MockedAIOConsumer({"bootstrap.servers": "localhost:9092"})
-            consumer._queue.append(MockedMessage("test-topic", 0, 0, []))
 
+            # MockedAIOConsumer wraps an inner MockConsumer, mirroring how
+            # the real AIOConsumer delegates to an inner Consumer internally.
+            # The inner Consumer is patched with wrap_poll below to simulate
+            # AutoInstrumentedConsumer — this is the only layer that should
+            # create spans; wrap_aio_poll must not produce a second set.
+            queue = [MockedMessage("test-topic", 0, 0, [])]
+            consumer = MockedAIOConsumer(
+                queue,
+                {"bootstrap.servers": "localhost:9092", "group.id": "g", "auto.offset.reset": "earliest"},
+            )
+
+            # Instrument the inner consumer's poll so spans are produced there.
+            inner = consumer._inner
+            raw_poll = inner.poll
+
+            def instrumented_inner_poll(timeout=None):
+                return ConfluentKafkaInstrumentor.wrap_poll(raw_poll, inner, tracer, (timeout,), {})
+
+            inner.poll = instrumented_inner_poll
+
+            # wrap_aio_poll is a pass-through; span is created by instrumented_inner_poll.
             msg = await instrumentation.wrap_aio_poll(
                 consumer.poll,
                 consumer,
@@ -526,6 +545,7 @@ async def _run():
             )
             self.assertIsNotNone(msg)
 
+            # Second (empty) poll causes the inner consumer to end the previous span.
             await instrumentation.wrap_aio_poll(
                 consumer.poll,
                 consumer,
@@ -535,7 +555,9 @@ async def _run():
             )
 
             spans = self.memory_exporter.get_finished_spans()
-            self.assertTrue(any(s.name == "test-topic process" for s in spans))
+            process_spans = [s for s in spans if s.name == "test-topic process"]
+            # Exactly ONE process span — no duplicates from the AIO wrapper layer.
+            self.assertEqual(len(process_spans), 1)
 
         asyncio.run(_run())
 
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
index b2fd638d59a..b02870d3e3c 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
@@ -9,6 +9,8 @@ class MockConsumer(Consumer):
     def __init__(self, queue, config):
         self._queue = queue
         self.config = config
+        self._current_consume_span = None
+        self._current_context_token = None
         super().__init__(config)
 
     def consume(self, num_messages=1, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
@@ -88,7 +90,7 @@ def __init__(self, config=None):
         self._queue = []
         self.config = config or {}
 
-    async def produce(self, topic, value=None, *args, **kwargs):
+    async def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
         msg = MockedMessage(
             topic=topic,
             partition=0,
@@ -102,15 +104,21 @@ async def produce(self, topic, value=None, *args, **kwargs):
 
 
 class MockedAIOConsumer:
-    def __init__(self, config=None):
-        self._queue = []
+    """Mock of confluent_kafka.AIOConsumer.
+
+    The real AIOConsumer delegates polling to an inner sync Consumer, which when
+    instrumented already creates spans via wrap_poll.  This mock mirrors that
+    structure so tests can verify that no duplicate spans are produced at the
+    AIO layer.
+    """
+
+    def __init__(self, queue, config=None):
+        self._inner = MockConsumer(queue, config or {})
         self.config = config or {}
-        self._current_consume_span = None
 
     async def poll(self, timeout=None):
-        if len(self._queue) > 0:
-            return self._queue.pop(0)
-        return None
+        # Delegate to the inner consumer, just like the real AIOConsumer does.
+        return self._inner.poll(timeout)
 
     async def close(self):
         pass

From ec0743645e42befc5c39d7835e072a12d2c3754a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Em=C3=ADdio?= <9735060+emdneto@users.noreply.github.com>
Date: Wed, 4 Nov 2026 10:00:16 -0300
Subject: [PATCH 06/10] Add new changelog entry for 5181

---
 .changelog/{5115.added => 5181.added} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename .changelog/{5115.added => 5181.added} (100%)

diff --git a/.changelog/5115.added b/.changelog/5181.added
similarity index 100%
rename from .changelog/5115.added
rename to .changelog/5181.added

From 43c269313841522359d1482d4611e9504ac1bbfa Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Fri, 6 Nov 2026 01:29:33 +0530
Subject: [PATCH 07/10] fix: re-attach AIO consumer span to async task context,
 add executor-backed test

---
 .../confluent_kafka/__init__.py               | 41 ++++++++++++++++---
 .../tests/test_instrumentation.py             | 28 ++++++++-----
 .../tests/utils.py                            | 19 +++++----
 3 files changed, 64 insertions(+), 24 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 9d28edd2a42..884294c684c 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -204,8 +204,9 @@ class AutoInstrumentedAIOConsumer(AIOConsumer):
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
             self.config = _capture_config(args, kwargs)
-            # Span lifecycle is managed by the inner Consumer layer, which is
-            # already instrumented via AutoInstrumentedConsumer.
+            # Token for the span re-attached to the caller's async task context.
+            # Managed by wrap_aio_poll / wrap_aio_close.
+            self._aio_context_token = None
 
         # Deliberately implemented so wrapt can intercept the async poll call
         async def poll(self, timeout=-1):
@@ -562,11 +563,39 @@ async def wrap_aio_produce(func, instance, tracer, args, kwargs):
 
     @staticmethod
     async def wrap_aio_poll(func, instance, tracer, args, kwargs):
-        # AIOConsumer delegates to an inner Consumer that is already instrumented
-        # via AutoInstrumentedConsumer. Creating spans here would produce duplicates,
-        # so this wrapper is intentionally a pass-through.
-        return await func(*args, **kwargs)
+        # Detach any span attached to the caller's async context from a
+        # previous poll call.  Keeping the token lifetime to a single poll
+        # prevents stale context if the caller never calls close().
+        if getattr(instance, "_aio_context_token", None):
+            context.detach(instance._aio_context_token)
+            instance._aio_context_token = None
+
+        # The real AIOConsumer runs Consumer.poll() inside a ThreadPoolExecutor
+        # via loop.run_in_executor().  Any context.attach() done in that worker
+        # thread is confined to the thread's context copy and is NOT visible in
+        # the caller's async task.  We therefore await first and then
+        # re-attach the span to the caller's async task context ourselves.
+        record = await func(*args, **kwargs)
+
+        if record:
+            # Read the span created by the instrumented inner Consumer on the
+            # worker thread and attach it to the caller's async task context.
+            inner = getattr(instance, "_consumer", None)
+            if inner is not None:
+                span = getattr(inner, "_current_consume_span", None)
+                if span is not None:
+                    # Clear the thread-level token so it is never detached
+                    # from a different thread or async context (which would
+                    # raise ValueError / silently corrupt the context chain).
+                    inner._current_context_token = None
+                    instance._aio_context_token = context.attach(trace.set_span_in_context(span))
+
+        return record
 
     @staticmethod
     async def wrap_aio_close(func, instance, args, kwargs):
+        # Detach any span still attached to the async task before closing.
+        if getattr(instance, "_aio_context_token", None):
+            context.detach(instance._aio_context_token)
+            instance._aio_context_token = None
         return await func(*args, **kwargs)
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
index fdd733a49f5..50280fd4a8c 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
@@ -3,6 +3,7 @@
 # pylint: disable=no-name-in-module,import-outside-toplevel
 import asyncio
 
+from opentelemetry import trace
 from opentelemetry.instrumentation.confluent_kafka import (
     AutoInstrumentedConsumer,
     AutoInstrumentedProducer,
@@ -515,19 +516,18 @@ async def _run():
             instrumentation = ConfluentKafkaInstrumentor()
             tracer = self.tracer_provider.get_tracer(__name__)
 
-            # MockedAIOConsumer wraps an inner MockConsumer, mirroring how
-            # the real AIOConsumer delegates to an inner Consumer internally.
-            # The inner Consumer is patched with wrap_poll below to simulate
-            # AutoInstrumentedConsumer — this is the only layer that should
-            # create spans; wrap_aio_poll must not produce a second set.
+            # MockedAIOConsumer dispatches the inner poll to a ThreadPoolExecutor,
+            # matching the real AIOConsumer's behaviour.  This exercises the path
+            # where the span is created on a worker thread and must be re-attached
+            # to the caller's async-task context by wrap_aio_poll.
             queue = [MockedMessage("test-topic", 0, 0, [])]
             consumer = MockedAIOConsumer(
                 queue,
                 {"bootstrap.servers": "localhost:9092", "group.id": "g", "auto.offset.reset": "earliest"},
             )
 
-            # Instrument the inner consumer's poll so spans are produced there.
-            inner = consumer._inner
+            # Instrument the inner consumer's poll to simulate AutoInstrumentedConsumer.
+            inner = consumer._consumer
             raw_poll = inner.poll
 
             def instrumented_inner_poll(timeout=None):
@@ -535,7 +535,8 @@ def instrumented_inner_poll(timeout=None):
 
             inner.poll = instrumented_inner_poll
 
-            # wrap_aio_poll is a pass-through; span is created by instrumented_inner_poll.
+            # First poll: inner Consumer creates a span on a worker thread;
+            # wrap_aio_poll re-attaches it to the caller's async task context.
             msg = await instrumentation.wrap_aio_poll(
                 consumer.poll,
                 consumer,
@@ -545,7 +546,12 @@ def instrumented_inner_poll(timeout=None):
             )
             self.assertIsNotNone(msg)
 
-            # Second (empty) poll causes the inner consumer to end the previous span.
+            # The span must now be current in the caller's async task.
+            active_span = trace.get_current_span()
+            self.assertTrue(active_span.is_recording())
+            self.assertEqual(active_span.name, "test-topic process")
+
+            # Second (empty) poll detaches the previous context and ends the span.
             await instrumentation.wrap_aio_poll(
                 consumer.poll,
                 consumer,
@@ -556,9 +562,11 @@ def instrumented_inner_poll(timeout=None):
 
             spans = self.memory_exporter.get_finished_spans()
             process_spans = [s for s in spans if s.name == "test-topic process"]
-            # Exactly ONE process span — no duplicates from the AIO wrapper layer.
+            # Exactly one process span — no duplicates from the AIO wrapper layer.
             self.assertEqual(len(process_spans), 1)
 
+            await consumer.close()
+
         asyncio.run(_run())
 
     def test_aio_instrument_and_uninstrument(self) -> None:
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
index b02870d3e3c..f0931721f7f 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
@@ -1,6 +1,8 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import asyncio
+import concurrent.futures
 
 from confluent_kafka import Consumer, Producer
 
@@ -106,19 +108,20 @@ async def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=
 class MockedAIOConsumer:
     """Mock of confluent_kafka.AIOConsumer.
 
-    The real AIOConsumer delegates polling to an inner sync Consumer, which when
-    instrumented already creates spans via wrap_poll.  This mock mirrors that
-    structure so tests can verify that no duplicate spans are produced at the
-    AIO layer.
+    Dispatches the inner Consumer poll to a ThreadPoolExecutor to replicate
+    the thread-pool behaviour of the real AIOConsumer, so that tests cover
+    the async-task context-management in wrap_aio_poll.
     """
 
     def __init__(self, queue, config=None):
-        self._inner = MockConsumer(queue, config or {})
+        self._consumer = MockConsumer(queue, config or {})
         self.config = config or {}
+        self._aio_context_token = None
+        self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
 
     async def poll(self, timeout=None):
-        # Delegate to the inner consumer, just like the real AIOConsumer does.
-        return self._inner.poll(timeout)
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(self._executor, self._consumer.poll, timeout)
 
     async def close(self):
-        pass
+        self._executor.shutdown(wait=False)

From 8967bfc9f6c6399280af8d70974137d6cef293b4 Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Fri, 6 Nov 2026 23:58:08 +0530
Subject: [PATCH 08/10] fix: suppress pylint false positives in AIO imports and
 tests

---
 .../instrumentation/confluent_kafka/__init__.py    | 14 +++++++-------
 .../tests/test_instrumentation.py                  |  2 +-
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 884294c684c..6b599f3cae6 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -99,17 +99,17 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ...
 
 from collections.abc import Collection
 
-import confluent_kafka
+import confluent_kafka  # pylint: disable=import-self
 import wrapt
-from confluent_kafka import Consumer, Producer
+from confluent_kafka import Consumer, Producer  # pylint: disable=import-self
 
 try:
-    import confluent_kafka.aio as _aio_module
-    from confluent_kafka.aio import AIOConsumer, AIOProducer
+    import confluent_kafka.aio as _aio_module  # pylint: disable=no-name-in-module
+    from confluent_kafka.aio import AIOConsumer, AIOProducer  # pylint: disable=no-name-in-module
 except ImportError:
     try:
-        import confluent_kafka.experimental.aio as _aio_module
-        from confluent_kafka.experimental.aio import (
+        import confluent_kafka.experimental.aio as _aio_module  # pylint: disable=no-name-in-module
+        from confluent_kafka.experimental.aio import (  # pylint: disable=no-name-in-module
             AIOConsumer,
             AIOProducer,
         )
@@ -192,7 +192,7 @@ def __init__(self, *args, **kwargs):
             self.config = _capture_config(args, kwargs)
 
         # Deliberately implemented so wrapt can intercept the async produce call
-        async def produce(self, topic, value=None, *args, **kwargs):
+        async def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
             return await super().produce(topic, value, *args, **kwargs)
 else:
     AutoInstrumentedAIOProducer = None
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
index 50280fd4a8c..415d04c7136 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
@@ -510,7 +510,7 @@ async def _run():
         asyncio.run(_run())
 
     def test_aio_poll(self) -> None:
-        from .utils import MockedAIOConsumer, MockedMessage  # noqa: PLC0415
+        from .utils import MockedAIOConsumer  # noqa: PLC0415
 
         async def _run():
             instrumentation = ConfluentKafkaInstrumentor()

From f9ca70ef0702db38b040dc0bb8afc67b54cd862e Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Mon, 9 Nov 2026 12:02:48 +0530
Subject: [PATCH 09/10] docs(confluent-kafka): add AIOProducer and AIOConsumer
 to nitpick exceptions

---
 docs/nitpick-exceptions.ini | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/docs/nitpick-exceptions.ini b/docs/nitpick-exceptions.ini
index 1d3a0d5dbad..d1e9009986a 100644
--- a/docs/nitpick-exceptions.ini
+++ b/docs/nitpick-exceptions.ini
@@ -15,6 +15,16 @@ py-class=
     opentelemetry.instrumentation.confluent_kafka.ProxiedConsumer
     opentelemetry.instrumentation.confluent_kafka.AutoInstrumentedAIOProducer
     opentelemetry.instrumentation.confluent_kafka.AutoInstrumentedAIOConsumer
+    confluent_kafka.aio.producer._AIOProducer.AIOProducer
+    confluent_kafka.aio._AIOConsumer.AIOConsumer
+    confluent_kafka.aio.AIOProducer
+    confluent_kafka.aio.AIOConsumer
+    confluent_kafka.experimental.aio.producer._AIOProducer.AIOProducer
+    confluent_kafka.experimental.aio._AIOConsumer.AIOConsumer
+    confluent_kafka.experimental.aio.AIOProducer
+    confluent_kafka.experimental.aio.AIOConsumer
+    AIOProducer
+    AIOConsumer
     opentelemetry.instrumentation.instrumentor.BaseInstrumentor
     ; - AwsXRayIdGenerator
     TextMapPropagator

From 29a8ecb6149c30b629480134dabe2cf0f32bbae4 Mon Sep 17 00:00:00 2001
From: Heisenbrick <ayushxmahato@gmail.com>
Date: Tue, 10 Nov 2026 22:06:40 +0530
Subject: [PATCH 10/10] fix(confluent-kafka): isolate worker context, capture
 config kwargs, and wrap aio consume

---
 .../confluent_kafka/__init__.py               | 65 +++++++++++--
 .../tests/test_instrumentation.py             | 96 +++++++++++++++++++
 .../tests/utils.py                            |  5 +
 3 files changed, 156 insertions(+), 10 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
index 22f502e597b..fccc881e397 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py
@@ -145,9 +145,10 @@ def _capture_config(args, kwargs):
     ``conf=`` kwarg, or (for Consumer) expanded as **kwargs."""
     if args and isinstance(args[0], dict):
         return args[0]
-    conf = kwargs.get("conf")
-    if isinstance(conf, dict):
-        return conf
+    for key in ("conf", "producer_conf", "consumer_conf"):
+        conf = kwargs.get(key)
+        if isinstance(conf, dict):
+            return conf
     # confluent_kafka.Consumer also supports Consumer(**conf) — in that case
     # the kwargs themselves are the config.
     if kwargs:
@@ -206,13 +207,23 @@ def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
             self.config = _capture_config(args, kwargs)
             # Token for the span re-attached to the caller's async task context.
-            # Managed by wrap_aio_poll / wrap_aio_close.
+            # Managed by wrap_aio_poll / wrap_aio_consume / wrap_aio_close.
             self._aio_context_token = None
+            inner = getattr(self, "_consumer", None)
+            if inner is not None:
+                try:
+                    inner._is_aio = True
+                except AttributeError:
+                    pass
 
         # Deliberately implemented so wrapt can intercept the async poll call
         async def poll(self, timeout=-1):
             return await super().poll(timeout)
 
+        # Deliberately implemented so wrapt can intercept the async consume call
+        async def consume(self, *args, **kwargs):  # pylint: disable=useless-super-delegation
+            return await super().consume(*args, **kwargs)
+
         # Deliberately implemented so wrapt can intercept the async close call
         async def close(self):
             return await super().close()
@@ -424,6 +435,9 @@ def _inner_wrap_aio_produce(func, instance, args, kwargs):
             def _inner_wrap_aio_poll(func, instance, args, kwargs):
                 return ConfluentKafkaInstrumentor.wrap_aio_poll(func, instance, self._tracer, args, kwargs)
 
+            def _inner_wrap_aio_consume(func, instance, args, kwargs):
+                return ConfluentKafkaInstrumentor.wrap_aio_consume(func, instance, self._tracer, args, kwargs)
+
             def _inner_wrap_aio_close(func, instance, args, kwargs):
                 return ConfluentKafkaInstrumentor.wrap_aio_close(func, instance, args, kwargs)
 
@@ -432,6 +446,11 @@ def _inner_wrap_aio_close(func, instance, args, kwargs):
                 "poll",
                 _inner_wrap_aio_poll,
             )
+            wrapt.wrap_function_wrapper(
+                AutoInstrumentedAIOConsumer,
+                "consume",
+                _inner_wrap_aio_consume,
+            )
             wrapt.wrap_function_wrapper(
                 AutoInstrumentedAIOConsumer,
                 "close",
@@ -459,6 +478,7 @@ def _uninstrument(self, **kwargs):
             unwrap(AutoInstrumentedAIOProducer, "produce")
         if AutoInstrumentedAIOConsumer is not None:
             unwrap(AutoInstrumentedAIOConsumer, "poll")
+            unwrap(AutoInstrumentedAIOConsumer, "consume")
             unwrap(AutoInstrumentedAIOConsumer, "close")
 
     @staticmethod
@@ -510,7 +530,10 @@ def wrap_poll(func, instance, tracer, args, kwargs):
                 "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER, attributes=receive_attributes
             ):
                 _create_new_consume_span(instance, tracer, [record], attributes)
-            instance._current_context_token = context.attach(trace.set_span_in_context(instance._current_consume_span))
+            if not getattr(instance, "_is_aio", False):
+                instance._current_context_token = context.attach(
+                    trace.set_span_in_context(instance._current_consume_span)
+                )
 
         return record
 
@@ -535,7 +558,10 @@ def wrap_consume(func, instance, tracer, args, kwargs):
                 "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER, attributes=receive_attributes
             ):
                 _create_new_consume_span(instance, tracer, records, attributes)
-            instance._current_context_token = context.attach(trace.set_span_in_context(instance._current_consume_span))
+            if not getattr(instance, "_is_aio", False):
+                instance._current_context_token = context.attach(
+                    trace.set_span_in_context(instance._current_consume_span)
+                )
 
         return records
 
@@ -561,6 +587,10 @@ async def wrap_aio_produce(func, instance, tracer, args, kwargs):
             bootstrap_servers=bootstrap_servers,
         )
         with tracer.start_as_current_span(name=span_name, kind=trace.SpanKind.PRODUCER, attributes=attributes):
+            # Note: confluent-kafka's AIOProducer does not currently support the
+            # headers keyword argument and raises NotImplementedError if headers are
+            # passed when none were provided. Trace context propagation via headers
+            # is therefore applied only when the caller explicitly passes headers.
             headers = KafkaPropertiesExtractor.extract_produce_headers(args, kwargs)
             if headers is not None:
                 propagate.inject(
@@ -593,14 +623,29 @@ async def wrap_aio_poll(func, instance, tracer, args, kwargs):
             if inner is not None:
                 span = getattr(inner, "_current_consume_span", None)
                 if span is not None:
-                    # Clear the thread-level token so it is never detached
-                    # from a different thread or async context (which would
-                    # raise ValueError / silently corrupt the context chain).
-                    inner._current_context_token = None
                     instance._aio_context_token = context.attach(trace.set_span_in_context(span))
 
         return record
 
+    @staticmethod
+    async def wrap_aio_consume(func, instance, tracer, args, kwargs):
+        # Detach any span attached to the caller's async context from a
+        # previous consume or poll call.
+        if getattr(instance, "_aio_context_token", None):
+            context.detach(instance._aio_context_token)
+            instance._aio_context_token = None
+
+        records = await func(*args, **kwargs)
+
+        if records and len(records) > 0:
+            inner = getattr(instance, "_consumer", None)
+            if inner is not None:
+                span = getattr(inner, "_current_consume_span", None)
+                if span is not None:
+                    instance._aio_context_token = context.attach(trace.set_span_in_context(span))
+
+        return records
+
     @staticmethod
     async def wrap_aio_close(func, instance, args, kwargs):
         # Detach any span still attached to the async task before closing.
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
index 2b6f71617a8..8a05a9d6185 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py
@@ -686,6 +686,11 @@ def instrumented_inner_poll(timeout=None):
             self.assertTrue(active_span.is_recording())
             self.assertEqual(active_span.name, "test-topic process")
 
+            # Verify worker thread context is not leaked to executor workers.
+            loop = asyncio.get_running_loop()
+            worker_span = await loop.run_in_executor(consumer._executor, trace.get_current_span)
+            self.assertFalse(worker_span.is_recording())
+
             # Second (empty) poll detaches the previous context and ends the span.
             await instrumentation.wrap_aio_poll(
                 consumer.poll,
@@ -700,10 +705,101 @@ def instrumented_inner_poll(timeout=None):
             # Exactly one process span — no duplicates from the AIO wrapper layer.
             self.assertEqual(len(process_spans), 1)
 
+            # Context detached in async task:
+            self.assertFalse(trace.get_current_span().is_recording())
+
             await consumer.close()
 
         asyncio.run(_run())
 
+    def test_aio_consume(self) -> None:
+        from .utils import MockedAIOConsumer  # noqa: PLC0415
+
+        async def _run():
+            instrumentation = ConfluentKafkaInstrumentor()
+            tracer = self.tracer_provider.get_tracer(__name__)
+
+            queue = [
+                MockedMessage("test-topic", 0, 0, []),
+                MockedMessage("test-topic", 0, 1, []),
+            ]
+            consumer = MockedAIOConsumer(
+                queue,
+                {"bootstrap.servers": "localhost:9092", "group.id": "g", "auto.offset.reset": "earliest"},
+            )
+
+            inner = consumer._consumer
+            raw_consume = inner.consume
+
+            def instrumented_inner_consume(*args, **kwargs):
+                return ConfluentKafkaInstrumentor.wrap_consume(raw_consume, inner, tracer, args, kwargs)
+
+            inner.consume = instrumented_inner_consume
+
+            # First batch consume
+            msgs = await instrumentation.wrap_aio_consume(
+                consumer.consume,
+                consumer,
+                tracer,
+                args=(2,),
+                kwargs={},
+            )
+            self.assertEqual(len(msgs), 2)
+
+            # The span must now be current in the caller's async task.
+            active_span = trace.get_current_span()
+            self.assertTrue(active_span.is_recording())
+            self.assertEqual(active_span.name, "test-topic process")
+
+            # Subsequent empty consume ends the previous span and detaches context.
+            await instrumentation.wrap_aio_consume(
+                consumer.consume,
+                consumer,
+                tracer,
+                args=(1,),
+                kwargs={},
+            )
+
+            spans = self.memory_exporter.get_finished_spans()
+            process_spans = [s for s in spans if s.name == "test-topic process"]
+            self.assertEqual(len(process_spans), 1)
+
+            # Context detached in async task:
+            self.assertFalse(trace.get_current_span().is_recording())
+
+            await consumer.close()
+
+        asyncio.run(_run())
+
+    def test_aio_config_kwargs(self) -> None:
+        from opentelemetry.instrumentation.confluent_kafka import (  # noqa: PLC0415
+            AutoInstrumentedAIOConsumer,
+            AutoInstrumentedAIOProducer,
+        )
+
+        async def _run():
+            instrumentor = ConfluentKafkaInstrumentor()
+            instrumentor.instrument()
+            try:
+                if AutoInstrumentedAIOProducer is not None:
+                    prod = AutoInstrumentedAIOProducer(producer_conf={"bootstrap.servers": "broker-p:9092"})
+                    self.assertEqual(prod.config.get("bootstrap.servers"), "broker-p:9092")
+                    await prod.close()
+
+                if AutoInstrumentedAIOConsumer is not None:
+                    cons = AutoInstrumentedAIOConsumer(
+                        consumer_conf={"bootstrap.servers": "broker-c:9092", "group.id": "g"}
+                    )
+                    self.assertEqual(cons.config.get("bootstrap.servers"), "broker-c:9092")
+                    inner = getattr(cons, "_consumer", None)
+                    if inner is not None:
+                        self.assertTrue(getattr(inner, "_is_aio", False))
+                    await cons.close()
+            finally:
+                instrumentor.uninstrument()
+
+        asyncio.run(_run())
+
     def test_aio_instrument_and_uninstrument(self) -> None:
         from opentelemetry.instrumentation.confluent_kafka import (  # noqa: PLC0415
             AutoInstrumentedAIOConsumer,
diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
index f0931721f7f..937f0c7fe20 100644
--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/utils.py
@@ -115,6 +115,7 @@ class MockedAIOConsumer:
 
     def __init__(self, queue, config=None):
         self._consumer = MockConsumer(queue, config or {})
+        self._consumer._is_aio = True
         self.config = config or {}
         self._aio_context_token = None
         self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
@@ -123,5 +124,9 @@ async def poll(self, timeout=None):
         loop = asyncio.get_running_loop()
         return await loop.run_in_executor(self._executor, self._consumer.poll, timeout)
 
+    async def consume(self, num_messages=1, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(self._executor, self._consumer.consume, num_messages, *args, **kwargs)
+
     async def close(self):
         self._executor.shutdown(wait=False)