fix(instrumentation-confluent-kafka): add __getattr__ delegation to ProxiedProducer and ProxiedConsumer

.changelog/4648.fixed

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-confluent-kafka`: forward unproxied methods to the wrapped producer/consumer

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

@@ -168,76 +168,78 @@ def close(self):  # pylint: disable=useless-super-delegation
         return super().close()
 
 
-class ProxiedProducer(Producer):
+class ProxiedProducer(wrapt.ObjectProxy):
+    # ObjectProxy transparently forwards every attribute and method not
+    # overridden below to the wrapped producer, so calls such as list_topics()
+    # or set_sasl_credentials() reach the underlying object instead of raising
+    # AttributeError / segfaulting on the C-extension type (see #4278).
     def __init__(self, producer: Producer, tracer: Tracer):
-        self._producer = producer
-        self._tracer = tracer
-        # Surface the wrapped producer's config (if any) so that
-        # KafkaPropertiesExtractor.extract_bootstrap_servers can read it
-        # through this proxy.
-        self.config = getattr(producer, "config", None)
-
-    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)
+        return ConfluentKafkaInstrumentor.wrap_produce(
+            self.__wrapped__.produce, self, self._self_tracer, args, new_kwargs
+        )
 
-    def original_producer(self):
-        return self._producer
+    def original_producer(self) -> Producer:
+        return self.__wrapped__
 
 
-class ProxiedConsumer(Consumer):
+class ProxiedConsumer(wrapt.ObjectProxy):
+    # See ProxiedProducer for the transparent-forwarding rationale.
     def __init__(self, consumer: Consumer, tracer: Tracer):
-        self._consumer = consumer
-        self._tracer = tracer
-        self._current_consume_span = None
-        self._current_context_token = None
-        # See ProxiedProducer.__init__ for rationale.
-        self.config = getattr(consumer, "config", None)
+        super().__init__(consumer)
+        self._self_tracer = tracer
+        self._self_current_consume_span = None
+        self._self_current_context_token = None
+
+    # The consume-tracking helpers in utils.py read and write these two
+    # attributes on the instance. Back them with ObjectProxy's `_self_`
+    # storage via properties so the state lives on the proxy rather than being
+    # forwarded onto the wrapped consumer, whose C-extension type does not
+    # accept arbitrary attributes.
+    @property
+    def _current_consume_span(self):
+        return self._self_current_consume_span
+
+    @_current_consume_span.setter
+    def _current_consume_span(self, value) -> None:
+        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) -> None:
+        self._self_current_context_token = value
 
     def close(self, *args, **kwargs):
-        return ConfluentKafkaInstrumentor.wrap_close(self._consumer.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)
+        return ConfluentKafkaInstrumentor.wrap_close(
+            self.__wrapped__.close, self, 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], {})
-
-    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)
+        return ConfluentKafkaInstrumentor.wrap_poll(
+            self.__wrapped__.poll, self, self._self_tracer, [timeout], {}
+        )
 
-    def original_consumer(self):
-        return self._consumer
+    def original_consumer(self) -> Consumer:
+        return self.__wrapped__
 
 
 class ConfluentKafkaInstrumentor(BaseInstrumentor):

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

@@ -41,7 +41,11 @@ def test_instrument_api(self) -> None:
         producer = Producer({"bootstrap.servers": "localhost:29092"})
         producer = instrumentation.instrument_producer(producer)
 
-        self.assertEqual(producer.__class__, ProxiedProducer)
+        # The proxies are wrapt.ObjectProxy subclasses, which deliberately
+        # report the *wrapped* object's __class__, so assert on the proxy type
+        # rather than comparing __class__ directly. Checks against the
+        # unwrapped class below are unaffected.
+        self.assertIsInstance(producer, ProxiedProducer)
 
         producer = instrumentation.uninstrument_producer(producer)
         self.assertEqual(producer.__class__, Producer)
@@ -55,7 +59,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)
@@ -69,7 +73,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)
@@ -116,7 +120,9 @@ def test_consumer_commit_method_exists(self) -> None:
         )
 
         consumer = instrumentation.instrument_consumer(consumer)
-        self.assertEqual(consumer.__class__, ProxiedConsumer)
+        # See test_instrument_api: ObjectProxy reports the wrapped __class__.
+        self.assertIsInstance(consumer, ProxiedConsumer)
+        # commit() is no longer declared on the proxy; ObjectProxy forwards it.
         self.assertTrue(hasattr(consumer, "commit"))
 
     def test_context_setter(self) -> None:
@@ -470,3 +476,46 @@ 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_proxied_producer_delegates_unproxied_methods(self) -> None:
+        """ProxiedProducer (a wrapt.ObjectProxy) must forward methods not
+        overridden on the proxy to the underlying producer, preventing
+        AttributeError or segfaults (see #4278)."""
+        from confluent_kafka import Producer  # noqa: PLC0415
+
+        instrumentation = ConfluentKafkaInstrumentor()
+        producer = MockedProducer([], {"bootstrap.servers": "localhost:9092"})
+        # A method that is *not* overridden on ProxiedProducer.
+        producer.custom_metadata = lambda: {"clusters": 1}
+        proxied = instrumentation.instrument_producer(producer)
+        self.assertIsInstance(proxied, ProxiedProducer)
+        # ObjectProxy exposes the wrapped object and keeps its identity.
+        self.assertIs(proxied.__wrapped__, producer)
+        self.assertIsInstance(proxied, Producer)
+        # Transparently forwarded, not raising AttributeError.
+        self.assertEqual(proxied.custom_metadata(), {"clusters": 1})
+
+    def test_proxied_consumer_delegates_unproxied_methods(self) -> None:
+        """ProxiedConsumer (a wrapt.ObjectProxy) must forward methods not
+        overridden on the proxy to the underlying consumer, preventing
+        AttributeError or segfaults (see #4278)."""
+        from confluent_kafka import Consumer  # noqa: PLC0415
+
+        instrumentation = ConfluentKafkaInstrumentor()
+        consumer = MockConsumer(
+            [],
+            {
+                "bootstrap.servers": "localhost:9092",
+                "group.id": "test-group",
+                "auto.offset.reset": "earliest",
+            },
+        )
+        # A method that is *not* overridden on ProxiedConsumer.
+        consumer.custom_assignment = lambda: []
+        proxied = instrumentation.instrument_consumer(consumer)
+        self.assertIsInstance(proxied, ProxiedConsumer)
+        # ObjectProxy exposes the wrapped object and keeps its identity.
+        self.assertIs(proxied.__wrapped__, consumer)
+        self.assertIsInstance(proxied, Consumer)
+        # Transparently forwarded, not raising AttributeError.
+        self.assertEqual(proxied.custom_assignment(), [])