From 27f7c2a37bc8d9c463a388702f434e8e4b4b2b76 Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 15:55:49 +0300 Subject: [PATCH 1/5] fix(aiokafka): close AIOKafkaProducer --- .../tests/test_instrumentation.py | 67 ++++++++++--------- 1 file changed, 35 insertions(+), 32 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py index 8905c2a68ed..62fc3f50f6d 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py @@ -498,37 +498,40 @@ async def test_send_and_wait(self) -> None: AIOKafkaInstrumentor().instrument(tracer_provider=self.tracer_provider) producer = await self.producer_factory() - add_message_mock: mock.AsyncMock = ( - producer._message_accumulator.add_message - ) - add_message_mock.side_effect = [mock.AsyncMock()(), mock.AsyncMock()()] - - tracer = self.tracer_provider.get_tracer(__name__) - with tracer.start_as_current_span("test_span") as span: - await producer.send_and_wait("topic_1", b"value_1") - - add_message_mock.assert_awaited_with( - TopicPartition(topic="topic_1", partition=1), - None, - b"value_1", - 40.0, - timestamp_ms=None, - headers=[("traceparent", mock.ANY)], - ) - assert ( - add_message_mock.call_args_list[0] - .kwargs["headers"][0][1] - .startswith( - f"00-{format_trace_id(span.get_span_context().trace_id)}-".encode() + try: + add_message_mock: mock.AsyncMock = ( + producer._message_accumulator.add_message + ) + add_message_mock.side_effect = [mock.AsyncMock()(), mock.AsyncMock()()] + + tracer = self.tracer_provider.get_tracer(__name__) + with tracer.start_as_current_span("test_span") as span: + await producer.send_and_wait("topic_1", b"value_1") + + add_message_mock.assert_awaited_with( + TopicPartition(topic="topic_1", partition=1), + None, + b"value_1", + 40.0, + timestamp_ms=None, + headers=[("traceparent", mock.ANY)], + ) + assert ( + add_message_mock.call_args_list[0] + .kwargs["headers"][0][1] + .startswith( + f"00-{format_trace_id(span.get_span_context().trace_id)}-".encode() + ) ) - ) - await producer.send_and_wait("topic_2", b"value_2") - add_message_mock.assert_awaited_with( - TopicPartition(topic="topic_2", partition=1), - None, - b"value_2", - 40.0, - timestamp_ms=None, - headers=[("traceparent", mock.ANY)], - ) + await producer.send_and_wait("topic_2", b"value_2") + add_message_mock.assert_awaited_with( + TopicPartition(topic="topic_2", partition=1), + None, + b"value_2", + 40.0, + timestamp_ms=None, + headers=[("traceparent", mock.ANY)], + ) + finally: + await producer.stop() From 82340b3db824c55ff02b775c70d10275d346ea8f Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 15:56:19 +0300 Subject: [PATCH 2/5] fix(aiokafka): use AsyncMock to fix unawaited coroutine warning --- .../opentelemetry-instrumentation-aiokafka/tests/test_utils.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index a7dcbe6dcfa..282df20802a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -256,7 +256,7 @@ async def test_wrap_getmany( extract_bootstrap_servers: mock.MagicMock, _enrich_getmany_poll_span: mock.MagicMock, _enrich_getmany_topic_span: mock.MagicMock, - _create_consumer_span: mock.MagicMock, + _create_consumer_span: mock.AsyncMock, extract: mock.MagicMock, ) -> None: tracer = mock.MagicMock() @@ -270,6 +270,7 @@ async def test_wrap_getmany( } ) kafka_consumer = mock.MagicMock() + _create_consumer_span.return_value = mock.MagicMock() wrapped_getmany = _wrap_getmany(tracer, consume_hook) records = await wrapped_getmany( From a33a5e60a673414e1b3257426284acb78dc96374 Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 16:18:01 +0300 Subject: [PATCH 3/5] docs(aiokafka): add CHANGELOG entry for test warning fixes --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c9c67be4bb..c52588be103 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ([#4302](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4302)) - `opentelemetry-instrumentation-grpc`: Fix bidirectional streaming RPCs raising `AttributeError: 'generator' object has no attribute 'add_done_callback'` ([#4259](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4259)) +- `opentelemetry-instrumentation-aiokafka`: Fix `Unclosed AIOKafkaProducer` warning in tests by properly stopping the producer, and fix `RuntimeWarning: coroutine was never awaited` by using `AsyncMock` for async mock targets ### Breaking changes From 1354a3d34984b0515657f6bdfcbbc78c5f81f76b Mon Sep 17 00:00:00 2001 From: dorlib Date: Thu, 2 Apr 2026 15:30:12 +0300 Subject: [PATCH 4/5] fix formatting --- .../tests/test_instrumentation.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py index 62fc3f50f6d..e074430252a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py @@ -502,7 +502,10 @@ async def test_send_and_wait(self) -> None: add_message_mock: mock.AsyncMock = ( producer._message_accumulator.add_message ) - add_message_mock.side_effect = [mock.AsyncMock()(), mock.AsyncMock()()] + add_message_mock.side_effect = [ + mock.AsyncMock()(), + mock.AsyncMock()(), + ] tracer = self.tracer_provider.get_tracer(__name__) with tracer.start_as_current_span("test_span") as span: From 40f30a5a686e74930b37342793305c4455b6521f Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Thu, 2 Apr 2026 17:57:22 +0200 Subject: [PATCH 5/5] Apply suggestion from @xrmx --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index af218b75eca..fb89ba90781 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,7 +40,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ([#4302](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4302)) - `opentelemetry-instrumentation-grpc`: Fix bidirectional streaming RPCs raising `AttributeError: 'generator' object has no attribute 'add_done_callback'` ([#4259](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4259)) -- `opentelemetry-instrumentation-aiokafka`: Fix `Unclosed AIOKafkaProducer` warning in tests by properly stopping the producer, and fix `RuntimeWarning: coroutine was never awaited` by using `AsyncMock` for async mock targets +- `opentelemetry-instrumentation-aiokafka`: fix `Unclosed AIOKafkaProducer` warning and `RuntimeWarning: coroutine was never awaited` in tests + ([#4384](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4384)) ### Breaking changes