confluent-kafka: add basic autoinstrumentation tests (#4266)

* confluent-kafka: test auto-instrumentation code path

* Add TODO to simpligy the instrumentation

* silence pylint

* Update __init__.py

Author: xrmx

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

@@ -277,6 +277,9 @@ def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
     def _instrument(self, **kwargs):
+        # TODO: should probably wrap methods directly instead of going through
+        # these classes. Hopefully it'll make the patching work if called after
+        # the original classes have already been imported,  #4270
         self._original_kafka_producer = confluent_kafka.Producer
         self._original_kafka_consumer = confluent_kafka.Consumer
 

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

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# pylint: disable=no-name-in-module
-
-from confluent_kafka import Consumer, Producer
+# pylint: disable=no-name-in-module,import-outside-toplevel
 
 from opentelemetry.instrumentation.confluent_kafka import (
+    AutoInstrumentedConsumer,
+    AutoInstrumentedProducer,
     ConfluentKafkaInstrumentor,
     ProxiedConsumer,
     ProxiedProducer,
@@ -41,6 +41,8 @@
 
 class TestConfluentKafka(TestBase):
     def test_instrument_api(self) -> None:
+        from confluent_kafka import Consumer, Producer  # noqa: PLC0415
+
         instrumentation = ConfluentKafkaInstrumentor()
 
         producer = Producer({"bootstrap.servers": "localhost:29092"})
@@ -51,16 +53,22 @@ def test_instrument_api(self) -> None:
         producer = instrumentation.uninstrument_producer(producer)
         self.assertEqual(producer.__class__, Producer)
 
-        producer = Producer({"bootstrap.servers": "localhost:29092"})
-        producer = instrumentation.instrument_producer(producer)
+        consumer = Consumer(
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            }
+        )
 
-        self.assertEqual(producer.__class__, ProxiedProducer)
+        consumer = instrumentation.instrument_consumer(consumer)
+        self.assertEqual(consumer.__class__, ProxiedConsumer)
 
-        producer = instrumentation.uninstrument_producer(producer)
-        self.assertEqual(producer.__class__, Producer)
+        consumer = instrumentation.uninstrument_consumer(consumer)
+        self.assertEqual(consumer.__class__, Consumer)
 
         consumer = Consumer(
-            {
+            **{
                 "bootstrap.servers": "localhost:29092",
                 "group.id": "mygroup",
                 "auto.offset.reset": "earliest",
@@ -73,7 +81,37 @@ def test_instrument_api(self) -> None:
         consumer = instrumentation.uninstrument_consumer(consumer)
         self.assertEqual(consumer.__class__, Consumer)
 
+    def test_instrument_api_with_instrument(self) -> None:
+        ConfluentKafkaInstrumentor().instrument()
+
+        from confluent_kafka import Consumer, Producer  # noqa: PLC0415
+
+        producer = Producer({"bootstrap.servers": "localhost:29092"})
+        self.assertEqual(producer.__class__, AutoInstrumentedProducer)
+
+        consumer = Consumer(
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            }
+        )
+        self.assertEqual(consumer.__class__, AutoInstrumentedConsumer)
+
+        consumer = Consumer(
+            **{
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            }
+        )
+        self.assertEqual(consumer.__class__, AutoInstrumentedConsumer)
+
+        ConfluentKafkaInstrumentor().uninstrument()
+
     def test_consumer_commit_method_exists(self) -> None:
+        from confluent_kafka import Consumer  # noqa: PLC0415
+
         instrumentation = ConfluentKafkaInstrumentor()
 
         consumer = Consumer(