We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix ConfluentKafkaInstrumentor usage
**Describe your environment** 
Docker image running `python:3.10.7-slim-bullseye` as base

**Steps to reproduce**
```python3
from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor

# Instrument kafka
ConfluentKafkaInstrumentor().instrument()
```

**What is the expected behavior?**
Instrumentation to work just by adding the lines above.

**What is the actual behavior?**

> ModuleNotFoundError: No module named 'opentelemetry.instrumentation.confluentkafka'

**Additional context**
The solution is to actually make the import like this:
```python3
from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 """
16 Instrument `confluent-kafka-python` to report instrumentation-confluent-kafka produced and consumed messages
17 
18 Usage
19 -----
20 
21 ..code:: python
22 
23     from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor
24     from confluent_kafka import Producer, Consumer
25 
26     # Instrument kafka
27     ConfluentKafkaInstrumentor().instrument()
28 
29     # report a span of type producer with the default settings
30     conf1 = {'bootstrap.servers': "localhost:9092"}
31     producer = Producer(conf1)
32     producer.produce('my-topic',b'raw_bytes')
33 
34     conf2 = {'bootstrap.servers': "localhost:9092",
35         'group.id': "foo",
36         'auto.offset.reset': 'smallest'}
37     # report a span of type consumer with the default settings
38     consumer = Consumer(conf2)
39     def basic_consume_loop(consumer, topics):
40         try:
41             consumer.subscribe(topics)
42             running = True
43             while running:
44                 msg = consumer.poll(timeout=1.0)
45                 if msg is None: continue
46 
47                 if msg.error():
48                     if msg.error().code() == KafkaError._PARTITION_EOF:
49                         # End of partition event
50                         sys.stderr.write(f"{msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}}\n")
51                     elif msg.error():
52                         raise KafkaException(msg.error())
53                 else:
54                     msg_process(msg)
55         finally:
56             # Close down consumer to commit final offsets.
57             consumer.close()
58 
59     basic_consume_loop(consumer, "my-topic")
60 
61 
62 The `_instrument` method accepts the following keyword args:
63 tracer_provider (TracerProvider) - an optional tracer provider
64 instrument_producer (Callable) - a function with extra user-defined logic to be performed before sending the message
65                           this function signature is:
66                           def instrument_producer(producer: Producer, tracer_provider=None)
67 instrument_consumer (Callable) - a function with extra user-defined logic to be performed after consuming a message
68                           this function signature is:
69                           def instrument_consumer(consumer: Consumer, tracer_provider=None)
70 for example:
71 .. code: python
72     from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor
73     from confluent_kafka import Producer, Consumer
74 
75     inst = ConfluentKafkaInstrumentor()
76 
77     p = confluent_kafka.Producer({'bootstrap.servers': 'localhost:29092'})
78     c = confluent_kafka.Consumer({
79         'bootstrap.servers': 'localhost:29092',
80         'group.id': 'mygroup',
81         'auto.offset.reset': 'earliest'
82     })
83 
84     # instrument confluent kafka with produce and consume hooks
85     p = inst.instrument_producer(p, tracer_provider)
86     c = inst.instrument_consumer(c, tracer_provider=tracer_provider)
87 
88 
89     # Using kafka as normal now will automatically generate spans,
90     # including user custom attributes added from the hooks
91     conf = {'bootstrap.servers': "localhost:9092"}
92     p.produce('my-topic',b'raw_bytes')
93     msg = c.poll()
94 
95 
96 API
97 ___
98 """
99 from typing import Collection
100 
101 import confluent_kafka
102 import wrapt
103 from confluent_kafka import Consumer, Producer
104 
105 from opentelemetry import context, propagate, trace
106 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
107 from opentelemetry.instrumentation.utils import unwrap
108 from opentelemetry.semconv.trace import MessagingOperationValues
109 from opentelemetry.trace import Link, SpanKind, Tracer
110 
111 from .package import _instruments
112 from .utils import (
113     KafkaPropertiesExtractor,
114     _enrich_span,
115     _get_span_name,
116     _kafka_getter,
117     _kafka_setter,
118 )
119 from .version import __version__
120 
121 
122 class AutoInstrumentedProducer(Producer):
123 
124     # This method is deliberately implemented in order to allow wrapt to wrap this function
125     def produce(
126         self, topic, value=None, *args, **kwargs
127     ):  # pylint: disable=keyword-arg-before-vararg,useless-super-delegation
128         super().produce(topic, value, *args, **kwargs)
129 
130 
131 class AutoInstrumentedConsumer(Consumer):
132     def __init__(self, config):
133         super().__init__(config)
134         self._current_consume_span = None
135 
136     # This method is deliberately implemented in order to allow wrapt to wrap this function
137     def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation
138         return super().poll(timeout)
139 
140 
141 class ProxiedProducer(Producer):
142     def __init__(self, producer: Producer, tracer: Tracer):
143         self._producer = producer
144         self._tracer = tracer
145 
146     def flush(self, timeout=-1):
147         self._producer.flush(timeout)
148 
149     def poll(self, timeout=-1):
150         self._producer.poll(timeout)
151 
152     def produce(
153         self, topic, value=None, *args, **kwargs
154     ):  # pylint: disable=keyword-arg-before-vararg
155         new_kwargs = kwargs.copy()
156         new_kwargs["topic"] = topic
157         new_kwargs["value"] = value
158 
159         return ConfluentKafkaInstrumentor.wrap_produce(
160             self._producer.produce, self, self._tracer, args, new_kwargs
161         )
162 
163     def original_producer(self):
164         return self._producer
165 
166 
167 class ProxiedConsumer(Consumer):
168     def __init__(self, consumer: Consumer, tracer: Tracer):
169         self._consumer = consumer
170         self._tracer = tracer
171         self._current_consume_span = None
172         self._current_context_token = None
173 
174     def committed(self, partitions, timeout=-1):
175         return self._consumer.committed(partitions, timeout)
176 
177     def consume(
178         self, num_messages=1, *args, **kwargs
179     ):  # pylint: disable=keyword-arg-before-vararg
180         return self._consumer.consume(num_messages, *args, **kwargs)
181 
182     def get_watermark_offsets(
183         self, partition, timeout=-1, *args, **kwargs
184     ):  # pylint: disable=keyword-arg-before-vararg
185         return self._consumer.get_watermark_offsets(
186             partition, timeout, *args, **kwargs
187         )
188 
189     def offsets_for_times(self, partitions, timeout=-1):
190         return self._consumer.offsets_for_times(partitions, timeout)
191 
192     def poll(self, timeout=-1):
193         return ConfluentKafkaInstrumentor.wrap_poll(
194             self._consumer.poll, self, self._tracer, [timeout], {}
195         )
196 
197     def subscribe(
198         self, topics, on_assign=lambda *args: None, *args, **kwargs
199     ):  # pylint: disable=keyword-arg-before-vararg
200         self._consumer.subscribe(topics, on_assign, *args, **kwargs)
201 
202     def original_consumer(self):
203         return self._consumer
204 
205 
206 class ConfluentKafkaInstrumentor(BaseInstrumentor):
207     """An instrumentor for confluent kafka module
208     See `BaseInstrumentor`
209     """
210 
211     # pylint: disable=attribute-defined-outside-init
212     @staticmethod
213     def instrument_producer(
214         producer: Producer, tracer_provider=None
215     ) -> ProxiedProducer:
216         tracer = trace.get_tracer(
217             __name__, __version__, tracer_provider=tracer_provider
218         )
219 
220         manual_producer = ProxiedProducer(producer, tracer)
221 
222         return manual_producer
223 
224     @staticmethod
225     def instrument_consumer(
226         consumer: Consumer, tracer_provider=None
227     ) -> ProxiedConsumer:
228         tracer = trace.get_tracer(
229             __name__, __version__, tracer_provider=tracer_provider
230         )
231 
232         manual_consumer = ProxiedConsumer(consumer, tracer)
233 
234         return manual_consumer
235 
236     @staticmethod
237     def uninstrument_producer(producer: Producer) -> Producer:
238         if isinstance(producer, ProxiedProducer):
239             return producer.original_producer()
240         return producer
241 
242     @staticmethod
243     def uninstrument_consumer(consumer: Consumer) -> Consumer:
244         if isinstance(consumer, ProxiedConsumer):
245             return consumer.original_consumer()
246         return consumer
247 
248     def instrumentation_dependencies(self) -> Collection[str]:
249         return _instruments
250 
251     def _instrument(self, **kwargs):
252         self._original_kafka_producer = confluent_kafka.Producer
253         self._original_kafka_consumer = confluent_kafka.Consumer
254 
255         confluent_kafka.Producer = AutoInstrumentedProducer
256         confluent_kafka.Consumer = AutoInstrumentedConsumer
257 
258         tracer_provider = kwargs.get("tracer_provider")
259         tracer = trace.get_tracer(
260             __name__, __version__, tracer_provider=tracer_provider
261         )
262 
263         self._tracer = tracer
264 
265         def _inner_wrap_produce(func, instance, args, kwargs):
266             return ConfluentKafkaInstrumentor.wrap_produce(
267                 func, instance, self._tracer, args, kwargs
268             )
269 
270         def _inner_wrap_poll(func, instance, args, kwargs):
271             return ConfluentKafkaInstrumentor.wrap_poll(
272                 func, instance, self._tracer, args, kwargs
273             )
274 
275         wrapt.wrap_function_wrapper(
276             AutoInstrumentedProducer,
277             "produce",
278             _inner_wrap_produce,
279         )
280 
281         wrapt.wrap_function_wrapper(
282             AutoInstrumentedConsumer,
283             "poll",
284             _inner_wrap_poll,
285         )
286 
287     def _uninstrument(self, **kwargs):
288         confluent_kafka.Producer = self._original_kafka_producer
289         confluent_kafka.Consumer = self._original_kafka_consumer
290 
291         unwrap(AutoInstrumentedProducer, "produce")
292         unwrap(AutoInstrumentedConsumer, "poll")
293 
294     @staticmethod
295     def wrap_produce(func, instance, tracer, args, kwargs):
296         topic = kwargs.get("topic")
297         if not topic:
298             topic = args[0]
299 
300         span_name = _get_span_name("send", topic)
301         with tracer.start_as_current_span(
302             name=span_name, kind=trace.SpanKind.PRODUCER
303         ) as span:
304             headers = KafkaPropertiesExtractor.extract_produce_headers(
305                 args, kwargs
306             )
307             if headers is None:
308                 headers = []
309                 kwargs["headers"] = headers
310 
311             topic = KafkaPropertiesExtractor.extract_produce_topic(args)
312             _enrich_span(
313                 span,
314                 topic,
315                 operation=MessagingOperationValues.RECEIVE,
316             )  # Replace
317             propagate.inject(
318                 headers,
319                 setter=_kafka_setter,
320             )
321             return func(*args, **kwargs)
322 
323     @staticmethod
324     def wrap_poll(func, instance, tracer, args, kwargs):
325         if instance._current_consume_span:
326             context.detach(instance._current_context_token)
327             instance._current_context_token = None
328             instance._current_consume_span.end()
329             instance._current_consume_span = None
330 
331         with tracer.start_as_current_span(
332             "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER
333         ):
334             record = func(*args, **kwargs)
335             if record:
336                 links = []
337                 ctx = propagate.extract(record.headers(), getter=_kafka_getter)
338                 if ctx:
339                     for item in ctx.values():
340                         if hasattr(item, "get_span_context"):
341                             links.append(Link(context=item.get_span_context()))
342 
343                 instance._current_consume_span = tracer.start_span(
344                     name=f"{record.topic()} process",
345                     links=links,
346                     kind=SpanKind.CONSUMER,
347                 )
348 
349                 _enrich_span(
350                     instance._current_consume_span,
351                     record.topic(),
352                     record.partition(),
353                     record.offset(),
354                     operation=MessagingOperationValues.PROCESS,
355                 )
356         instance._current_context_token = context.attach(
357             trace.set_span_in_context(instance._current_consume_span)
358         )
359 
360         return record
361
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

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
--- 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
@@ -20,7 +20,7 @@
 
 ..code:: python
 
-    from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor
+    from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor
     from confluent_kafka import Producer, Consumer
 
     # Instrument kafka
@@ -69,7 +69,7 @@
                           def instrument_consumer(consumer: Consumer, tracer_provider=None)
 for example:
 .. code: python
-    from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor
+    from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor
     from confluent_kafka import Producer, Consumer
 
     inst = ConfluentKafkaInstrumentor()

{"golden_diff": "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\n--- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py\n+++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py\n@@ -20,7 +20,7 @@\n \n ..code:: python\n \n-    from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor\n+    from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor\n     from confluent_kafka import Producer, Consumer\n \n     # Instrument kafka\n@@ -69,7 +69,7 @@\n                           def instrument_consumer(consumer: Consumer, tracer_provider=None)\n for example:\n .. code: python\n-    from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor\n+    from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor\n     from confluent_kafka import Producer, Consumer\n \n     inst = ConfluentKafkaInstrumentor()\n", "issue": "Fix ConfluentKafkaInstrumentor usage\n**Describe your environment** \r\nDocker image running `python:3.10.7-slim-bullseye` as base\r\n\r\n**Steps to reproduce**\r\n```python3\r\nfrom opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor\r\n\r\n# Instrument kafka\r\nConfluentKafkaInstrumentor().instrument()\r\n```\r\n\r\n**What is the expected behavior?**\r\nInstrumentation to work just by adding the lines above.\r\n\r\n**What is the actual behavior?**\r\n\r\n> ModuleNotFoundError: No module named 'opentelemetry.instrumentation.confluentkafka'\r\n\r\n**Additional context**\r\nThe solution is to actually make the import like this:\r\n```python3\r\nfrom opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor\r\n```\r\n\n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nInstrument `confluent-kafka-python` to report instrumentation-confluent-kafka produced and consumed messages\n\nUsage\n-----\n\n..code:: python\n\n    from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor\n    from confluent_kafka import Producer, Consumer\n\n    # Instrument kafka\n    ConfluentKafkaInstrumentor().instrument()\n\n    # report a span of type producer with the default settings\n    conf1 = {'bootstrap.servers': \"localhost:9092\"}\n    producer = Producer(conf1)\n    producer.produce('my-topic',b'raw_bytes')\n\n    conf2 = {'bootstrap.servers': \"localhost:9092\",\n        'group.id': \"foo\",\n        'auto.offset.reset': 'smallest'}\n    # report a span of type consumer with the default settings\n    consumer = Consumer(conf2)\n    def basic_consume_loop(consumer, topics):\n        try:\n            consumer.subscribe(topics)\n            running = True\n            while running:\n                msg = consumer.poll(timeout=1.0)\n                if msg is None: continue\n\n                if msg.error():\n                    if msg.error().code() == KafkaError._PARTITION_EOF:\n                        # End of partition event\n                        sys.stderr.write(f\"{msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}}\\n\")\n                    elif msg.error():\n                        raise KafkaException(msg.error())\n                else:\n                    msg_process(msg)\n        finally:\n            # Close down consumer to commit final offsets.\n            consumer.close()\n\n    basic_consume_loop(consumer, \"my-topic\")\n\n\nThe `_instrument` method accepts the following keyword args:\ntracer_provider (TracerProvider) - an optional tracer provider\ninstrument_producer (Callable) - a function with extra user-defined logic to be performed before sending the message\n                          this function signature is:\n                          def instrument_producer(producer: Producer, tracer_provider=None)\ninstrument_consumer (Callable) - a function with extra user-defined logic to be performed after consuming a message\n                          this function signature is:\n                          def instrument_consumer(consumer: Consumer, tracer_provider=None)\nfor example:\n.. code: python\n    from opentelemetry.instrumentation.confluentkafka import ConfluentKafkaInstrumentor\n    from confluent_kafka import Producer, Consumer\n\n    inst = ConfluentKafkaInstrumentor()\n\n    p = confluent_kafka.Producer({'bootstrap.servers': 'localhost:29092'})\n    c = confluent_kafka.Consumer({\n        'bootstrap.servers': 'localhost:29092',\n        'group.id': 'mygroup',\n        'auto.offset.reset': 'earliest'\n    })\n\n    # instrument confluent kafka with produce and consume hooks\n    p = inst.instrument_producer(p, tracer_provider)\n    c = inst.instrument_consumer(c, tracer_provider=tracer_provider)\n\n\n    # Using kafka as normal now will automatically generate spans,\n    # including user custom attributes added from the hooks\n    conf = {'bootstrap.servers': \"localhost:9092\"}\n    p.produce('my-topic',b'raw_bytes')\n    msg = c.poll()\n\n\nAPI\n___\n\"\"\"\nfrom typing import Collection\n\nimport confluent_kafka\nimport wrapt\nfrom confluent_kafka import Consumer, Producer\n\nfrom opentelemetry import context, propagate, trace\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.instrumentation.utils import unwrap\nfrom opentelemetry.semconv.trace import MessagingOperationValues\nfrom opentelemetry.trace import Link, SpanKind, Tracer\n\nfrom .package import _instruments\nfrom .utils import (\n    KafkaPropertiesExtractor,\n    _enrich_span,\n    _get_span_name,\n    _kafka_getter,\n    _kafka_setter,\n)\nfrom .version import __version__\n\n\nclass AutoInstrumentedProducer(Producer):\n\n    # This method is deliberately implemented in order to allow wrapt to wrap this function\n    def produce(\n        self, topic, value=None, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg,useless-super-delegation\n        super().produce(topic, value, *args, **kwargs)\n\n\nclass AutoInstrumentedConsumer(Consumer):\n    def __init__(self, config):\n        super().__init__(config)\n        self._current_consume_span = None\n\n    # This method is deliberately implemented in order to allow wrapt to wrap this function\n    def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation\n        return super().poll(timeout)\n\n\nclass ProxiedProducer(Producer):\n    def __init__(self, producer: Producer, tracer: Tracer):\n        self._producer = producer\n        self._tracer = tracer\n\n    def flush(self, timeout=-1):\n        self._producer.flush(timeout)\n\n    def poll(self, timeout=-1):\n        self._producer.poll(timeout)\n\n    def produce(\n        self, topic, value=None, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        new_kwargs = kwargs.copy()\n        new_kwargs[\"topic\"] = topic\n        new_kwargs[\"value\"] = value\n\n        return ConfluentKafkaInstrumentor.wrap_produce(\n            self._producer.produce, self, self._tracer, args, new_kwargs\n        )\n\n    def original_producer(self):\n        return self._producer\n\n\nclass ProxiedConsumer(Consumer):\n    def __init__(self, consumer: Consumer, tracer: Tracer):\n        self._consumer = consumer\n        self._tracer = tracer\n        self._current_consume_span = None\n        self._current_context_token = None\n\n    def committed(self, partitions, timeout=-1):\n        return self._consumer.committed(partitions, timeout)\n\n    def consume(\n        self, num_messages=1, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        return self._consumer.consume(num_messages, *args, **kwargs)\n\n    def get_watermark_offsets(\n        self, partition, timeout=-1, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        return self._consumer.get_watermark_offsets(\n            partition, timeout, *args, **kwargs\n        )\n\n    def offsets_for_times(self, partitions, timeout=-1):\n        return self._consumer.offsets_for_times(partitions, timeout)\n\n    def poll(self, timeout=-1):\n        return ConfluentKafkaInstrumentor.wrap_poll(\n            self._consumer.poll, self, self._tracer, [timeout], {}\n        )\n\n    def subscribe(\n        self, topics, on_assign=lambda *args: None, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        self._consumer.subscribe(topics, on_assign, *args, **kwargs)\n\n    def original_consumer(self):\n        return self._consumer\n\n\nclass ConfluentKafkaInstrumentor(BaseInstrumentor):\n    \"\"\"An instrumentor for confluent kafka module\n    See `BaseInstrumentor`\n    \"\"\"\n\n    # pylint: disable=attribute-defined-outside-init\n    @staticmethod\n    def instrument_producer(\n        producer: Producer, tracer_provider=None\n    ) -> ProxiedProducer:\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n\n        manual_producer = ProxiedProducer(producer, tracer)\n\n        return manual_producer\n\n    @staticmethod\n    def instrument_consumer(\n        consumer: Consumer, tracer_provider=None\n    ) -> ProxiedConsumer:\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n\n        manual_consumer = ProxiedConsumer(consumer, tracer)\n\n        return manual_consumer\n\n    @staticmethod\n    def uninstrument_producer(producer: Producer) -> Producer:\n        if isinstance(producer, ProxiedProducer):\n            return producer.original_producer()\n        return producer\n\n    @staticmethod\n    def uninstrument_consumer(consumer: Consumer) -> Consumer:\n        if isinstance(consumer, ProxiedConsumer):\n            return consumer.original_consumer()\n        return consumer\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\n    def _instrument(self, **kwargs):\n        self._original_kafka_producer = confluent_kafka.Producer\n        self._original_kafka_consumer = confluent_kafka.Consumer\n\n        confluent_kafka.Producer = AutoInstrumentedProducer\n        confluent_kafka.Consumer = AutoInstrumentedConsumer\n\n        tracer_provider = kwargs.get(\"tracer_provider\")\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n\n        self._tracer = tracer\n\n        def _inner_wrap_produce(func, instance, args, kwargs):\n            return ConfluentKafkaInstrumentor.wrap_produce(\n                func, instance, self._tracer, args, kwargs\n            )\n\n        def _inner_wrap_poll(func, instance, args, kwargs):\n            return ConfluentKafkaInstrumentor.wrap_poll(\n                func, instance, self._tracer, args, kwargs\n            )\n\n        wrapt.wrap_function_wrapper(\n            AutoInstrumentedProducer,\n            \"produce\",\n            _inner_wrap_produce,\n        )\n\n        wrapt.wrap_function_wrapper(\n            AutoInstrumentedConsumer,\n            \"poll\",\n            _inner_wrap_poll,\n        )\n\n    def _uninstrument(self, **kwargs):\n        confluent_kafka.Producer = self._original_kafka_producer\n        confluent_kafka.Consumer = self._original_kafka_consumer\n\n        unwrap(AutoInstrumentedProducer, \"produce\")\n        unwrap(AutoInstrumentedConsumer, \"poll\")\n\n    @staticmethod\n    def wrap_produce(func, instance, tracer, args, kwargs):\n        topic = kwargs.get(\"topic\")\n        if not topic:\n            topic = args[0]\n\n        span_name = _get_span_name(\"send\", topic)\n        with tracer.start_as_current_span(\n            name=span_name, kind=trace.SpanKind.PRODUCER\n        ) as span:\n            headers = KafkaPropertiesExtractor.extract_produce_headers(\n                args, kwargs\n            )\n            if headers is None:\n                headers = []\n                kwargs[\"headers\"] = headers\n\n            topic = KafkaPropertiesExtractor.extract_produce_topic(args)\n            _enrich_span(\n                span,\n                topic,\n                operation=MessagingOperationValues.RECEIVE,\n            )  # Replace\n            propagate.inject(\n                headers,\n                setter=_kafka_setter,\n            )\n            return func(*args, **kwargs)\n\n    @staticmethod\n    def wrap_poll(func, instance, tracer, args, kwargs):\n        if instance._current_consume_span:\n            context.detach(instance._current_context_token)\n            instance._current_context_token = None\n            instance._current_consume_span.end()\n            instance._current_consume_span = None\n\n        with tracer.start_as_current_span(\n            \"recv\", end_on_exit=True, kind=trace.SpanKind.CONSUMER\n        ):\n            record = func(*args, **kwargs)\n            if record:\n                links = []\n                ctx = propagate.extract(record.headers(), getter=_kafka_getter)\n                if ctx:\n                    for item in ctx.values():\n                        if hasattr(item, \"get_span_context\"):\n                            links.append(Link(context=item.get_span_context()))\n\n                instance._current_consume_span = tracer.start_span(\n                    name=f\"{record.topic()} process\",\n                    links=links,\n                    kind=SpanKind.CONSUMER,\n                )\n\n                _enrich_span(\n                    instance._current_consume_span,\n                    record.topic(),\n                    record.partition(),\n                    record.offset(),\n                    operation=MessagingOperationValues.PROCESS,\n                )\n        instance._current_context_token = context.attach(\n            trace.set_span_in_context(instance._current_consume_span)\n        )\n\n        return record\n", "path": "instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"\nInstrument `confluent-kafka-python` to report instrumentation-confluent-kafka produced and consumed messages\n\nUsage\n-----\n\n..code:: python\n\n    from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor\n    from confluent_kafka import Producer, Consumer\n\n    # Instrument kafka\n    ConfluentKafkaInstrumentor().instrument()\n\n    # report a span of type producer with the default settings\n    conf1 = {'bootstrap.servers': \"localhost:9092\"}\n    producer = Producer(conf1)\n    producer.produce('my-topic',b'raw_bytes')\n\n    conf2 = {'bootstrap.servers': \"localhost:9092\",\n        'group.id': \"foo\",\n        'auto.offset.reset': 'smallest'}\n    # report a span of type consumer with the default settings\n    consumer = Consumer(conf2)\n    def basic_consume_loop(consumer, topics):\n        try:\n            consumer.subscribe(topics)\n            running = True\n            while running:\n                msg = consumer.poll(timeout=1.0)\n                if msg is None: continue\n\n                if msg.error():\n                    if msg.error().code() == KafkaError._PARTITION_EOF:\n                        # End of partition event\n                        sys.stderr.write(f\"{msg.topic()} [{msg.partition()}] reached end at offset {msg.offset()}}\\n\")\n                    elif msg.error():\n                        raise KafkaException(msg.error())\n                else:\n                    msg_process(msg)\n        finally:\n            # Close down consumer to commit final offsets.\n            consumer.close()\n\n    basic_consume_loop(consumer, \"my-topic\")\n\n\nThe `_instrument` method accepts the following keyword args:\ntracer_provider (TracerProvider) - an optional tracer provider\ninstrument_producer (Callable) - a function with extra user-defined logic to be performed before sending the message\n                          this function signature is:\n                          def instrument_producer(producer: Producer, tracer_provider=None)\ninstrument_consumer (Callable) - a function with extra user-defined logic to be performed after consuming a message\n                          this function signature is:\n                          def instrument_consumer(consumer: Consumer, tracer_provider=None)\nfor example:\n.. code: python\n    from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor\n    from confluent_kafka import Producer, Consumer\n\n    inst = ConfluentKafkaInstrumentor()\n\n    p = confluent_kafka.Producer({'bootstrap.servers': 'localhost:29092'})\n    c = confluent_kafka.Consumer({\n        'bootstrap.servers': 'localhost:29092',\n        'group.id': 'mygroup',\n        'auto.offset.reset': 'earliest'\n    })\n\n    # instrument confluent kafka with produce and consume hooks\n    p = inst.instrument_producer(p, tracer_provider)\n    c = inst.instrument_consumer(c, tracer_provider=tracer_provider)\n\n\n    # Using kafka as normal now will automatically generate spans,\n    # including user custom attributes added from the hooks\n    conf = {'bootstrap.servers': \"localhost:9092\"}\n    p.produce('my-topic',b'raw_bytes')\n    msg = c.poll()\n\n\nAPI\n___\n\"\"\"\nfrom typing import Collection\n\nimport confluent_kafka\nimport wrapt\nfrom confluent_kafka import Consumer, Producer\n\nfrom opentelemetry import context, propagate, trace\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.instrumentation.utils import unwrap\nfrom opentelemetry.semconv.trace import MessagingOperationValues\nfrom opentelemetry.trace import Link, SpanKind, Tracer\n\nfrom .package import _instruments\nfrom .utils import (\n    KafkaPropertiesExtractor,\n    _enrich_span,\n    _get_span_name,\n    _kafka_getter,\n    _kafka_setter,\n)\nfrom .version import __version__\n\n\nclass AutoInstrumentedProducer(Producer):\n\n    # This method is deliberately implemented in order to allow wrapt to wrap this function\n    def produce(\n        self, topic, value=None, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg,useless-super-delegation\n        super().produce(topic, value, *args, **kwargs)\n\n\nclass AutoInstrumentedConsumer(Consumer):\n    def __init__(self, config):\n        super().__init__(config)\n        self._current_consume_span = None\n\n    # This method is deliberately implemented in order to allow wrapt to wrap this function\n    def poll(self, timeout=-1):  # pylint: disable=useless-super-delegation\n        return super().poll(timeout)\n\n\nclass ProxiedProducer(Producer):\n    def __init__(self, producer: Producer, tracer: Tracer):\n        self._producer = producer\n        self._tracer = tracer\n\n    def flush(self, timeout=-1):\n        self._producer.flush(timeout)\n\n    def poll(self, timeout=-1):\n        self._producer.poll(timeout)\n\n    def produce(\n        self, topic, value=None, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        new_kwargs = kwargs.copy()\n        new_kwargs[\"topic\"] = topic\n        new_kwargs[\"value\"] = value\n\n        return ConfluentKafkaInstrumentor.wrap_produce(\n            self._producer.produce, self, self._tracer, args, new_kwargs\n        )\n\n    def original_producer(self):\n        return self._producer\n\n\nclass ProxiedConsumer(Consumer):\n    def __init__(self, consumer: Consumer, tracer: Tracer):\n        self._consumer = consumer\n        self._tracer = tracer\n        self._current_consume_span = None\n        self._current_context_token = None\n\n    def committed(self, partitions, timeout=-1):\n        return self._consumer.committed(partitions, timeout)\n\n    def consume(\n        self, num_messages=1, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        return self._consumer.consume(num_messages, *args, **kwargs)\n\n    def get_watermark_offsets(\n        self, partition, timeout=-1, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        return self._consumer.get_watermark_offsets(\n            partition, timeout, *args, **kwargs\n        )\n\n    def offsets_for_times(self, partitions, timeout=-1):\n        return self._consumer.offsets_for_times(partitions, timeout)\n\n    def poll(self, timeout=-1):\n        return ConfluentKafkaInstrumentor.wrap_poll(\n            self._consumer.poll, self, self._tracer, [timeout], {}\n        )\n\n    def subscribe(\n        self, topics, on_assign=lambda *args: None, *args, **kwargs\n    ):  # pylint: disable=keyword-arg-before-vararg\n        self._consumer.subscribe(topics, on_assign, *args, **kwargs)\n\n    def original_consumer(self):\n        return self._consumer\n\n\nclass ConfluentKafkaInstrumentor(BaseInstrumentor):\n    \"\"\"An instrumentor for confluent kafka module\n    See `BaseInstrumentor`\n    \"\"\"\n\n    # pylint: disable=attribute-defined-outside-init\n    @staticmethod\n    def instrument_producer(\n        producer: Producer, tracer_provider=None\n    ) -> ProxiedProducer:\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n\n        manual_producer = ProxiedProducer(producer, tracer)\n\n        return manual_producer\n\n    @staticmethod\n    def instrument_consumer(\n        consumer: Consumer, tracer_provider=None\n    ) -> ProxiedConsumer:\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n\n        manual_consumer = ProxiedConsumer(consumer, tracer)\n\n        return manual_consumer\n\n    @staticmethod\n    def uninstrument_producer(producer: Producer) -> Producer:\n        if isinstance(producer, ProxiedProducer):\n            return producer.original_producer()\n        return producer\n\n    @staticmethod\n    def uninstrument_consumer(consumer: Consumer) -> Consumer:\n        if isinstance(consumer, ProxiedConsumer):\n            return consumer.original_consumer()\n        return consumer\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\n    def _instrument(self, **kwargs):\n        self._original_kafka_producer = confluent_kafka.Producer\n        self._original_kafka_consumer = confluent_kafka.Consumer\n\n        confluent_kafka.Producer = AutoInstrumentedProducer\n        confluent_kafka.Consumer = AutoInstrumentedConsumer\n\n        tracer_provider = kwargs.get(\"tracer_provider\")\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n\n        self._tracer = tracer\n\n        def _inner_wrap_produce(func, instance, args, kwargs):\n            return ConfluentKafkaInstrumentor.wrap_produce(\n                func, instance, self._tracer, args, kwargs\n            )\n\n        def _inner_wrap_poll(func, instance, args, kwargs):\n            return ConfluentKafkaInstrumentor.wrap_poll(\n                func, instance, self._tracer, args, kwargs\n            )\n\n        wrapt.wrap_function_wrapper(\n            AutoInstrumentedProducer,\n            \"produce\",\n            _inner_wrap_produce,\n        )\n\n        wrapt.wrap_function_wrapper(\n            AutoInstrumentedConsumer,\n            \"poll\",\n            _inner_wrap_poll,\n        )\n\n    def _uninstrument(self, **kwargs):\n        confluent_kafka.Producer = self._original_kafka_producer\n        confluent_kafka.Consumer = self._original_kafka_consumer\n\n        unwrap(AutoInstrumentedProducer, \"produce\")\n        unwrap(AutoInstrumentedConsumer, \"poll\")\n\n    @staticmethod\n    def wrap_produce(func, instance, tracer, args, kwargs):\n        topic = kwargs.get(\"topic\")\n        if not topic:\n            topic = args[0]\n\n        span_name = _get_span_name(\"send\", topic)\n        with tracer.start_as_current_span(\n            name=span_name, kind=trace.SpanKind.PRODUCER\n        ) as span:\n            headers = KafkaPropertiesExtractor.extract_produce_headers(\n                args, kwargs\n            )\n            if headers is None:\n                headers = []\n                kwargs[\"headers\"] = headers\n\n            topic = KafkaPropertiesExtractor.extract_produce_topic(args)\n            _enrich_span(\n                span,\n                topic,\n                operation=MessagingOperationValues.RECEIVE,\n            )  # Replace\n            propagate.inject(\n                headers,\n                setter=_kafka_setter,\n            )\n            return func(*args, **kwargs)\n\n    @staticmethod\n    def wrap_poll(func, instance, tracer, args, kwargs):\n        if instance._current_consume_span:\n            context.detach(instance._current_context_token)\n            instance._current_context_token = None\n            instance._current_consume_span.end()\n            instance._current_consume_span = None\n\n        with tracer.start_as_current_span(\n            \"recv\", end_on_exit=True, kind=trace.SpanKind.CONSUMER\n        ):\n            record = func(*args, **kwargs)\n            if record:\n                links = []\n                ctx = propagate.extract(record.headers(), getter=_kafka_getter)\n                if ctx:\n                    for item in ctx.values():\n                        if hasattr(item, \"get_span_context\"):\n                            links.append(Link(context=item.get_span_context()))\n\n                instance._current_consume_span = tracer.start_span(\n                    name=f\"{record.topic()} process\",\n                    links=links,\n                    kind=SpanKind.CONSUMER,\n                )\n\n                _enrich_span(\n                    instance._current_consume_span,\n                    record.topic(),\n                    record.partition(),\n                    record.offset(),\n                    operation=MessagingOperationValues.PROCESS,\n                )\n        instance._current_context_token = context.attach(\n            trace.set_span_in_context(instance._current_consume_span)\n        )\n\n        return record\n", "path": "instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py"}]}