opentelemetry-instrumentation-aiokafka: Broken with aiokafka 0.13 — _serialize signature change causes argument misalignment #4378

Describe your environment

OS: any
Python version: 3.13, 3.14
Package version: opentelemetry-instrumentation-aiokafka 0.60b1 (and current main)
aiokafka version: 0.13.0

What happened?

When using opentelemetry-instrumentation-aiokafka with aiokafka 0.13, the
instrumentation's _extract_send_partition() function fails silently on every
send() call. The function calls instance._serialize(topic, key, value),
but aiokafka 0.13 changed the signature of AIOKafkaProducer._serialize from
(self, topic, key, value) to (self, key, value, headers).

This causes the arguments to shift:

• topic is passed as key
• key (often None) is passed as value
• value (the actual payload) is passed as headers

When key serializers or value serializers are configured, they receive the wrong
arguments. In our case, the value serializer received None instead of the
actual Pydantic model, causing AttributeError: 'NoneType' object has no attribute 'model_dump'.

Steps to Reproduce

1. Install aiokafka 0.13.0 and opentelemetry-instrumentation-aiokafka
2. Create an AIOKafkaProducer with a custom value serializer
3. Enable OTel auto-instrumentation (e.g., opentelemetry-instrument)
4. Call producer.send(topic, value=my_object, key=None)

Expected Result

_extract_send_partition calls the value serializer with the actual value and
the key serializer with the actual key, correctly extracting the partition.

Actual Result

The value serializer receives None (the key argument), causing serialization failure.
_extract_send_partition returns None on every call.
When using custom serializers (e.g., Avro/Pydantic), this produces errors like:

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

Additional context

The root cause is in utils.py _extract_send_partition():

key_bytes, value_bytes = cast(
    "tuple[bytes | None, bytes | None]",
    instance._serialize(topic, key, value),
)

aiokafka 0.12 signature: _serialize(self, topic, key, value) → returns (key_bytes, value_bytes)
aiokafka 0.13 signature: _serialize(self, key, value, headers) → topic fills key, key fills value, value fills headers

The _serialize method is internal and its signature is not part of aiokafka's public API. The fix should call _key_serializer and _value_serializer directly, which are stable across versions.

aiokafka 0.13 changelog: https://github.com/aio-libs/aiokafka/releases/tag/v0.13.0

Would you like to implement a fix?

Yes, PR Opened #4379