diff --git a/CHANGELOG.md b/CHANGELOG.md
index 257510fec09..a96dc5ef7da 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -40,6 +40,10 @@ 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)`
+  ([#4379](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4379))
 
 ### Breaking changes
 
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/test-requirements.txt b/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt
index 460e58e0911..c29f54a732c 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/test-requirements.txt
@@ -1,4 +1,5 @@
-aiokafka==0.11.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 8905c2a68ed..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(
@@ -126,7 +131,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()
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 (