From 2f1a87f2584c8ccfe0191d73c5e4a921b2d50075 Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 14:27:55 +0300 Subject: [PATCH 1/6] tests(aiokafka): bump aiokafka 0.13 --- .../test-requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt b/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt index 460e58e0911..58de309991a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt @@ -1,4 +1,4 @@ -aiokafka==0.11.0 +aiokafka==0.13.0 pytest==7.4.4 -e opentelemetry-instrumentation -e instrumentation/opentelemetry-instrumentation-aiokafka From cf1759672e0a5505687f35ac40790dd91fc9fb77 Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 14:29:03 +0300 Subject: [PATCH 2/6] tests(aiokafka): remove removed api_version argument from AIOKafkaProducer --- .../tests/test_instrumentation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py index 8905c2a68ed..bfafa7c8c96 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py @@ -126,7 +126,7 @@ async def consumer_factory(**consumer_kwargs: Any) -> AIOKafkaConsumer: @staticmethod async def producer_factory() -> AIOKafkaProducer: - producer = AIOKafkaProducer(api_version="1.0") + producer = AIOKafkaProducer() producer.client._wait_on_metadata = mock.AsyncMock() producer.client.bootstrap = mock.AsyncMock() From a90265f4ed31a75cc886e6e75af60cc157937635 Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 14:32:14 +0300 Subject: [PATCH 3/6] fix(aiokafka): call serializers directly instead of _serialize --- .../instrumentation/aiokafka/utils.py | 14 +++++++++++--- .../tests/test_utils.py | 3 ++- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index 36bfb29efa4..8533a988589 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -162,9 +162,17 @@ async def _extract_send_partition( key = _extract_send_key(args, kwargs) value = _extract_send_value(args, kwargs) partition = _extract_argument("partition", 3, None, args, kwargs) - key_bytes, value_bytes = cast( - "tuple[bytes | None, bytes | None]", - instance._serialize(topic, key, value), # type: ignore[reportUnknownMemberType] + key_bytes = cast( + "bytes | None", + instance._key_serializer(key) # type: ignore[reportUnknownMemberType] + if instance._key_serializer # type: ignore[reportUnknownMemberType] + else key, + ) + value_bytes = cast( + "bytes | None", + instance._value_serializer(value) # type: ignore[reportUnknownMemberType] + if instance._value_serializer # type: ignore[reportUnknownMemberType] + else value, ) valid_types = (bytes, bytearray, memoryview, type(None)) if ( diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index a7dcbe6dcfa..5c3532692bb 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -370,7 +370,8 @@ async def test_create_consumer_span( async def test_kafka_properties_extractor(self): aiokafka_instance_mock = mock.Mock() - aiokafka_instance_mock._serialize.return_value = None, None + aiokafka_instance_mock._key_serializer = None + aiokafka_instance_mock._value_serializer = None aiokafka_instance_mock._partition.return_value = "partition" aiokafka_instance_mock.client._wait_on_metadata = mock.AsyncMock() assert ( From 65276c3f05f520fe15a595cc194c8685f90c71af Mon Sep 17 00:00:00 2001 From: dorlib Date: Wed, 1 Apr 2026 14:32:27 +0300 Subject: [PATCH 4/6] chore: add CHANGELOG entry for aiokafka 0.13 compatibility fix --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c9c67be4bb..f2a9ed41a3f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,9 @@ 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 compatibility with aiokafka 0.13 by calling + `_key_serializer`/`_value_serializer` directly instead of the internal `_serialize` method + whose signature changed in 0.13 from `(topic, key, value)` to `(key, value, headers)` ### Breaking changes From d066c443599830f3a3f6c4984323d781c54880a8 Mon Sep 17 00:00:00 2001 From: dorlib Date: Thu, 2 Apr 2026 14:19:57 +0300 Subject: [PATCH 5/6] tests(aiokafka): align tests to support python 3.9 --- .../test-requirements.txt | 3 ++- .../tests/test_instrumentation.py | 7 ++++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt b/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt index 58de309991a..c29f54a732c 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt @@ -1,4 +1,5 @@ -aiokafka==0.13.0 +aiokafka==0.13.0; python_version >= "3.10" +aiokafka==0.12.0; python_version < "3.10" pytest==7.4.4 -e opentelemetry-instrumentation -e instrumentation/opentelemetry-instrumentation-aiokafka diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py index bfafa7c8c96..10f63d62196 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py @@ -14,9 +14,10 @@ from __future__ import annotations +import sys import uuid from typing import Any, Sequence, cast -from unittest import IsolatedAsyncioTestCase, TestCase, mock +from unittest import IsolatedAsyncioTestCase, TestCase, mock, skipIf import aiokafka from aiokafka import ( @@ -63,6 +64,10 @@ def test_instrument_api(self) -> None: ) +@skipIf( + sys.version_info < (3, 10), + "aiokafka >= 0.13 requires Python 3.10+", +) class TestAIOKafkaInstrumentation(TestBase, IsolatedAsyncioTestCase): @staticmethod def consumer_record_factory( From 2ff5be92b586795d3aad548ebbb5a0ef290f508e Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Thu, 2 Apr 2026 18:02:12 +0200 Subject: [PATCH 6/6] Apply suggestion from @xrmx --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 427b247b413..a96dc5ef7da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `opentelemetry-instrumentation-aiokafka`: Fix compatibility with aiokafka 0.13 by calling `_key_serializer`/`_value_serializer` directly instead of the internal `_serialize` method whose signature changed in 0.13 from `(topic, key, value)` to `(key, value, headers)` + ([#4379](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4379)) ### Breaking changes