From 474a5867b5b232a068f5c4626a2489f5107ef493 Mon Sep 17 00:00:00 2001 From: stark256-spec Date: Tue, 2 Jun 2026 15:09:01 -0500 Subject: [PATCH 1/2] fix(instrumentation-confluent-kafka): add __getattr__ delegation to ProxiedProducer and ProxiedConsumer ProxiedProducer and ProxiedConsumer only forward an explicit list of methods to the underlying confluent-kafka objects. Calling any method outside that list (e.g. list_topics(), assignment(), memberid(), set_sasl_credentials()) raises AttributeError on pure-Python mocks or causes a segmentation fault on real confluent-kafka C-extension objects, because the C-extension Producer/Consumer base classes don't implement standard Python attribute lookup. Add __getattr__ to both proxy classes so that any attribute or method lookup not satisfied by an explicitly defined proxy method is forwarded to the underlying _producer/_consumer via getattr(). This is the standard Python delegation pattern and has no impact on methods that are already explicitly proxied. Fixes #4278 Assisted-by: Claude Sonnet 4.5 --- .../confluent_kafka/__init__.py | 14 ++++++++ .../tests/test_instrumentation.py | 33 +++++++++++++++++++ 2 files changed, 47 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 08aa4c3c7d..ee94ba495f 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 @@ -195,6 +195,13 @@ def produce(self, topic, value=None, *args, **kwargs): # pylint: disable=keywor self._producer.produce, self, self._tracer, args, new_kwargs ) + def __getattr__(self, name: str): + # Delegate any attribute or method not explicitly proxied to the + # underlying producer (e.g. list_topics(), set_sasl_credentials()). + # Without this, callers get AttributeError or — for confluent-kafka + # C-extension objects — a segmentation fault. + return getattr(self._producer, name) + def original_producer(self): return self._producer @@ -244,6 +251,13 @@ def poll(self, timeout=-1): 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 __getattr__(self, name: str): + # Delegate any attribute or method not explicitly proxied to the + # underlying consumer (e.g. assignment(), list_topics(), + # memberid()). Without this, callers get AttributeError or — for + # confluent-kafka C-extension objects — a segmentation fault. + return getattr(self._consumer, name) + def original_consumer(self): return self._consumer diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index 8252d4ffb4..389a8cc1f4 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -482,3 +482,36 @@ def test_consumer_sets_bootstrap_servers_attributes(self) -> None: ) 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.__getattr__ must forward methods not explicitly + defined on the proxy to the underlying producer, preventing + AttributeError or segfaults (see #4278).""" + instrumentation = ConfluentKafkaInstrumentor() + producer = MockedProducer([], {"bootstrap.servers": "localhost:9092"}) + # Add a sentinel method that is *not* explicitly defined on ProxiedProducer + producer.custom_metadata = lambda: {"clusters": 1} + proxied = instrumentation.instrument_producer(producer) + self.assertIsInstance(proxied, ProxiedProducer) + # Must be forwarded via __getattr__, not raise AttributeError + self.assertEqual(proxied.custom_metadata(), {"clusters": 1}) + + def test_proxied_consumer_delegates_unproxied_methods(self) -> None: + """ProxiedConsumer.__getattr__ must forward methods not explicitly + defined on the proxy to the underlying consumer, preventing + AttributeError or segfaults (see #4278).""" + instrumentation = ConfluentKafkaInstrumentor() + consumer = MockConsumer( + [], + { + "bootstrap.servers": "localhost:9092", + "group.id": "test-group", + "auto.offset.reset": "earliest", + }, + ) + # Add a sentinel method that is *not* explicitly defined on ProxiedConsumer + consumer.custom_assignment = lambda: [] + proxied = instrumentation.instrument_consumer(consumer) + self.assertIsInstance(proxied, ProxiedConsumer) + # Must be forwarded via __getattr__, not raise AttributeError + self.assertEqual(proxied.custom_assignment(), []) From ce417da844f975fc42809bdb55f69aad1f9dc008 Mon Sep 17 00:00:00 2001 From: stark256-spec Date: Thu, 16 Jul 2026 22:40:52 -0500 Subject: [PATCH 2/2] fix(instrumentation-confluent-kafka): use wrapt.ObjectProxy for proxy classes ProxiedProducer and ProxiedConsumer forwarded only an explicit list of methods to the underlying confluent-kafka objects, so any method outside that list (list_topics(), assignment(), memberid(), set_sasl_credentials()) raised AttributeError on pure-Python mocks or segfaulted on the real C-extension objects. Base both proxies on wrapt.ObjectProxy, which transparently forwards every attribute and method that is not explicitly overridden. This removes the hand-maintained pass-through list (flush, poll, purge, committed, commit, get_watermark_offsets, offsets_for_times, subscribe) along with the need for manual __getattr__ delegation, leaving only the instrumented methods (produce, poll, consume, close). Proxy-local state uses ObjectProxy's _self_ storage. The two attributes the consume-tracking helpers in utils.py read and write on the instance (_current_consume_span, _current_context_token) are exposed as properties backed by that storage, so the state stays on the proxy instead of being forwarded onto the wrapped C-extension client. Tests that asserted the exact proxy class now use assertIsInstance, since ObjectProxy intentionally reports the wrapped object's __class__. --- .changelog/4648.fixed | 1 + .../confluent_kafka/__init__.py | 106 +++++++----------- .../tests/test_instrumentation.py | 40 +++++-- 3 files changed, 72 insertions(+), 75 deletions(-) create mode 100644 .changelog/4648.fixed diff --git a/.changelog/4648.fixed b/.changelog/4648.fixed new file mode 100644 index 0000000000..a744d9de2a --- /dev/null +++ b/.changelog/4648.fixed @@ -0,0 +1 @@ +`opentelemetry-instrumentation-confluent-kafka`: forward unproxied methods to the wrapped producer/consumer 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 ee94ba495f..3c24e442a8 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 @@ -168,23 +168,14 @@ 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() @@ -192,74 +183,63 @@ def produce(self, topic, value=None, *args, **kwargs): # pylint: disable=keywor 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 __getattr__(self, name: str): - # Delegate any attribute or method not explicitly proxied to the - # underlying producer (e.g. list_topics(), set_sasl_credentials()). - # Without this, callers get AttributeError or — for confluent-kafka - # C-extension objects — a segmentation fault. - return getattr(self._producer, name) - - 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 + 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 __getattr__(self, name: str): - # Delegate any attribute or method not explicitly proxied to the - # underlying consumer (e.g. assignment(), list_topics(), - # memberid()). Without this, callers get AttributeError or — for - # confluent-kafka C-extension objects — a segmentation fault. - return getattr(self._consumer, name) - - def original_consumer(self): - return self._consumer + def original_consumer(self) -> Consumer: + return self.__wrapped__ class ConfluentKafkaInstrumentor(BaseInstrumentor): diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index 389a8cc1f4..1e16b33c64 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/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: @@ -484,22 +490,29 @@ def test_consumer_sets_bootstrap_servers_attributes(self) -> None: self.assertEqual(process_span.attributes[SERVER_PORT], 9092) def test_proxied_producer_delegates_unproxied_methods(self) -> None: - """ProxiedProducer.__getattr__ must forward methods not explicitly - defined on the proxy to the underlying producer, preventing + """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"}) - # Add a sentinel method that is *not* explicitly defined on ProxiedProducer + # A method that is *not* overridden on ProxiedProducer. producer.custom_metadata = lambda: {"clusters": 1} proxied = instrumentation.instrument_producer(producer) self.assertIsInstance(proxied, ProxiedProducer) - # Must be forwarded via __getattr__, not raise AttributeError + # 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.__getattr__ must forward methods not explicitly - defined on the proxy to the underlying consumer, preventing + """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( [], @@ -509,9 +522,12 @@ def test_proxied_consumer_delegates_unproxied_methods(self) -> None: "auto.offset.reset": "earliest", }, ) - # Add a sentinel method that is *not* explicitly defined on ProxiedConsumer + # A method that is *not* overridden on ProxiedConsumer. consumer.custom_assignment = lambda: [] proxied = instrumentation.instrument_consumer(consumer) self.assertIsInstance(proxied, ProxiedConsumer) - # Must be forwarded via __getattr__, not raise AttributeError + # 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(), [])