Fix/aiokafka 0.13 compatibility

[dorlib]

Description

_extract_send_partition() in the aiokafka instrumentation calls instance._serialize(topic, key, value) to serialize the key and value before partition assignment. In aiokafka 0.13, the internal _serialize method changed its signature from (self, topic, key, value) to (self, key, value, headers), causing the arguments to shift:

• topic (a string) is passed where key is expected
• key (often None) is passed where value is expected
• value (the actual payload) is passed where headers is expected

This causes custom serializers to receive wrong arguments. For example, a value serializer expecting a Pydantic model receives None instead, producing:

AttributeError: 'NoneType' object has no attribute 'model_dump'

The fix replaces the _serialize() call with direct calls to _key_serializer and _value_serializer, which are stable across aiokafka versions:

# Before (breaks on aiokafka 0.13):
key_bytes, value_bytes = cast(
    "tuple[bytes | None, bytes | None]",
    instance._serialize(topic, key, value),
)

# After (works on both 0.12 and 0.13):
key_bytes = instance._key_serializer(key) if instance._key_serializer else key
value_bytes = instance._value_serializer(value) if instance._value_serializer else value

api_version parameter has been removed from producer client in aiokafka 0.13
link

Fixes #4378

Type of change

Please delete options that are not relevant.

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• This change requires a documentation update

How Has This Been Tested?

• Existing unit tests updated (test_kafka_properties_extractor now mocks_key_serializer/_value_serializer instead of the removed _serialize call)
• New regression test added (test_extract_send_partition_with_custom_serializers) that verifies the value serializer receives the actual value and the key serializer receives the actual key. this test fails on main and passes with the fix
• Manually verified in a production-like environment with aiokafka 0.13.0, OTel auto-instrumentation enabled, and custom Avro/Pydantic serializers. clean logs, correct partition assignment, messages produced and consumed successfully
• All 18 existing tests pass: tox -e py3-test-instrumentation-aiokafka

Does This PR Require a Core Repo Change?

• Yes. - Link to PR:
• No.

Checklist:

• Followed the style guidelines of this project
• Changelogs have been updated
• Unit tests have been added
• Documentation has been updated

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: dorlib / name: dorlib (2f1a87f, 65276c3, a90265f, cf17596, d066c44)
• ✅ login: emdneto / name: Emídio Neto (5fa8dc9)
• ✅ login: xrmx / name: Riccardo Magliocchetti (2ff5be9, fb5b675)

[tammy-baylis-swi]

Thank you for this fix!

I'm a bit concerned that regular testing didn't catch this, I guess because they run on aiokafka==0.11.0.

A bit brittle, but what do others (@open-telemetry/opentelemetry-python-contrib-approvers) think of adjusting the test coverage like the sqlalchemy instrumentor does? The tox.ini points to multiple test-requirements files, e.g. -1 and 2.

[emdneto]

LGTM

[tammy-baylis-swi]

Lgtm too! Thank you