fix(aiokafka): call serializers directly instead of _serialize

Author: dorlib

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 (

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 (