fix #4278: use wrapt.ObjectProxy to preserve isinstance() compatibility

Replace __getattr__ composition with wrapt.ObjectProxy so that ProxiedProducer
and ProxiedConsumer remain valid isinstance() targets for the wrapped types.
Internal span-tracking state uses _self_ prefix (wrapt convention) exposed via
properties to keep utils.py interface unchanged.

Author: JMGalvao

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

@@ -151,84 +151,68 @@ def close(self):  # pylint: disable=useless-super-delegation
         return super().close()
 
 
-class ProxiedProducer:
+class ProxiedProducer(wrapt.ObjectProxy):
     def __init__(self, producer: Producer, tracer: Tracer):
-        self._producer = producer
-        self._tracer = tracer
-
-    def flush(self, timeout=-1):
-        return self._producer.flush(timeout)
-
-    def poll(self, timeout=-1):
-        return self._producer.poll(timeout)
-
-    def purge(self, in_queue=True, in_flight=True, blocking=True):
-        self._producer.purge(in_queue, in_flight, blocking)
+        super().__init__(producer)
+        self._self_tracer = tracer
 
     def produce(self, topic, value=None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
         new_kwargs = kwargs.copy()
         new_kwargs["topic"] = topic
         new_kwargs["value"] = value
 
         return ConfluentKafkaInstrumentor.wrap_produce(
-            self._producer.produce, self, self._tracer, args, new_kwargs
+            self.__wrapped__.produce, self, self._self_tracer, args, new_kwargs
         )
 
     def original_producer(self):
-        return self._producer
+        return self.__wrapped__
 
-    def __getattr__(self, name):
-        return getattr(self._producer, name)
 
-
-class ProxiedConsumer:
+class ProxiedConsumer(wrapt.ObjectProxy):
     def __init__(self, consumer: Consumer, tracer: Tracer):
-        self._consumer = consumer
-        self._tracer = tracer
-        self._current_consume_span = None
-        self._current_context_token = None
+        super().__init__(consumer)
+        self._self_tracer = tracer
+        self._self_current_consume_span = None
+        self._self_current_context_token = None
+
+    @property
+    def _current_consume_span(self):
+        return self._self_current_consume_span
+
+    @_current_consume_span.setter
+    def _current_consume_span(self, value):
+        self._self_current_consume_span = value
+
+    @property
+    def _current_context_token(self):
+        return self._self_current_context_token
+
+    @_current_context_token.setter
+    def _current_context_token(self, value):
+        self._self_current_context_token = value
 
     def close(self, *args, **kwargs):
         return ConfluentKafkaInstrumentor.wrap_close(
-            self._consumer.close, self, args, kwargs
+            self.__wrapped__.close, self, args, kwargs
         )
 
-    def committed(self, partitions, timeout=-1):
-        return self._consumer.committed(partitions, timeout)
-
-    def commit(self, *args, **kwargs):
-        return self._consumer.commit(*args, **kwargs)
-
     def consume(self, *args, **kwargs):
         return ConfluentKafkaInstrumentor.wrap_consume(
-            self._consumer.consume,
+            self.__wrapped__.consume,
             self,
-            self._tracer,
+            self._self_tracer,
             args,
             kwargs,
         )
 
-    def get_watermark_offsets(self, partition, timeout=-1, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
-        return self._consumer.get_watermark_offsets(
-            partition, timeout, *args, **kwargs
-        )
-
-    def offsets_for_times(self, partitions, timeout=-1):
-        return self._consumer.offsets_for_times(partitions, timeout)
-
     def poll(self, timeout=-1):
         return ConfluentKafkaInstrumentor.wrap_poll(
-            self._consumer.poll, self, self._tracer, [timeout], {}
+            self.__wrapped__.poll, self, self._self_tracer, [timeout], {}
         )
 
-    def subscribe(self, topics, on_assign=lambda *args: None, *args, **kwargs):  # pylint: disable=keyword-arg-before-vararg
-        self._consumer.subscribe(topics, on_assign, *args, **kwargs)
-
     def original_consumer(self):
-        return self._consumer
-
-    def __getattr__(self, name):
-        return getattr(self._consumer, name)
+        return self.__wrapped__
 
 
 class ConfluentKafkaInstrumentor(BaseInstrumentor):

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

@@ -48,7 +48,8 @@ def test_instrument_api(self) -> None:
         producer = Producer({"bootstrap.servers": "localhost:29092"})
         producer = instrumentation.instrument_producer(producer)
 
-        self.assertEqual(producer.__class__, ProxiedProducer)
+        self.assertIsInstance(producer, ProxiedProducer)
+        self.assertIsInstance(producer, Producer)
 
         producer = instrumentation.uninstrument_producer(producer)
         self.assertEqual(producer.__class__, Producer)
@@ -62,7 +63,8 @@ def test_instrument_api(self) -> None:
         )
 
         consumer = instrumentation.instrument_consumer(consumer)
-        self.assertEqual(consumer.__class__, ProxiedConsumer)
+        self.assertIsInstance(consumer, ProxiedConsumer)
+        self.assertIsInstance(consumer, Consumer)
 
         consumer = instrumentation.uninstrument_consumer(consumer)
         self.assertEqual(consumer.__class__, Consumer)
@@ -76,7 +78,7 @@ def test_instrument_api(self) -> None:
         )
 
         consumer = instrumentation.instrument_consumer(consumer)
-        self.assertEqual(consumer.__class__, ProxiedConsumer)
+        self.assertIsInstance(consumer, ProxiedConsumer)
 
         consumer = instrumentation.uninstrument_consumer(consumer)
         self.assertEqual(consumer.__class__, Consumer)
@@ -123,7 +125,8 @@ def test_consumer_commit_method_exists(self) -> None:
         )
 
         consumer = instrumentation.instrument_consumer(consumer)
-        self.assertEqual(consumer.__class__, ProxiedConsumer)
+        self.assertIsInstance(consumer, ProxiedConsumer)
+        self.assertIsInstance(consumer, Consumer)
         self.assertTrue(hasattr(consumer, "commit"))
 
     def test_context_setter(self) -> None:
@@ -465,6 +468,7 @@ def test_proxied_producer_delegates_undefined_methods(
         )
 
         proxied = instrumentation.instrument_producer(producer)
+        self.assertIsInstance(proxied, MockedProducer)
         self.assertEqual(proxied.list_topics(), "producer_topics")
 
     def test_proxied_consumer_delegates_undefined_methods(
@@ -485,5 +489,6 @@ def test_proxied_consumer_delegates_undefined_methods(
         )
 
         proxied = instrumentation.instrument_consumer(consumer)
+        self.assertIsInstance(proxied, MockConsumer)
         self.assertEqual(proxied.list_topics(), "consumer_topics")
         self.assertEqual(proxied.assignment(), [])