diff --git a/CHANGELOG.md b/CHANGELOG.md
index a96dc5ef7da..419a8a7c63d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -40,6 +40,8 @@ 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 `Unclosed AIOKafkaProducer` warning and `RuntimeWarning: coroutine was never awaited` in tests
+  ([#4384](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4384)) 
 - `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)`
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py
index 10f63d62196..646232da356 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py
@@ -503,37 +503,43 @@ async def test_send_and_wait(self) -> None:
         AIOKafkaInstrumentor().instrument(tracer_provider=self.tracer_provider)
 
         producer = await self.producer_factory()
-        add_message_mock: mock.AsyncMock = (
-            producer._message_accumulator.add_message
-        )
-        add_message_mock.side_effect = [mock.AsyncMock()(), mock.AsyncMock()()]
-
-        tracer = self.tracer_provider.get_tracer(__name__)
-        with tracer.start_as_current_span("test_span") as span:
-            await producer.send_and_wait("topic_1", b"value_1")
+        try:
+            add_message_mock: mock.AsyncMock = (
+                producer._message_accumulator.add_message
+            )
+            add_message_mock.side_effect = [
+                mock.AsyncMock()(),
+                mock.AsyncMock()(),
+            ]
 
-        add_message_mock.assert_awaited_with(
-            TopicPartition(topic="topic_1", partition=1),
-            None,
-            b"value_1",
-            40.0,
-            timestamp_ms=None,
-            headers=[("traceparent", mock.ANY)],
-        )
-        assert (
-            add_message_mock.call_args_list[0]
-            .kwargs["headers"][0][1]
-            .startswith(
-                f"00-{format_trace_id(span.get_span_context().trace_id)}-".encode()
+            tracer = self.tracer_provider.get_tracer(__name__)
+            with tracer.start_as_current_span("test_span") as span:
+                await producer.send_and_wait("topic_1", b"value_1")
+
+            add_message_mock.assert_awaited_with(
+                TopicPartition(topic="topic_1", partition=1),
+                None,
+                b"value_1",
+                40.0,
+                timestamp_ms=None,
+                headers=[("traceparent", mock.ANY)],
+            )
+            assert (
+                add_message_mock.call_args_list[0]
+                .kwargs["headers"][0][1]
+                .startswith(
+                    f"00-{format_trace_id(span.get_span_context().trace_id)}-".encode()
+                )
             )
-        )
 
-        await producer.send_and_wait("topic_2", b"value_2")
-        add_message_mock.assert_awaited_with(
-            TopicPartition(topic="topic_2", partition=1),
-            None,
-            b"value_2",
-            40.0,
-            timestamp_ms=None,
-            headers=[("traceparent", mock.ANY)],
-        )
+            await producer.send_and_wait("topic_2", b"value_2")
+            add_message_mock.assert_awaited_with(
+                TopicPartition(topic="topic_2", partition=1),
+                None,
+                b"value_2",
+                40.0,
+                timestamp_ms=None,
+                headers=[("traceparent", mock.ANY)],
+            )
+        finally:
+            await producer.stop()
diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
index 5c3532692bb..14e6d36178a 100644
--- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
+++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py
@@ -256,7 +256,7 @@ async def test_wrap_getmany(
         extract_bootstrap_servers: mock.MagicMock,
         _enrich_getmany_poll_span: mock.MagicMock,
         _enrich_getmany_topic_span: mock.MagicMock,
-        _create_consumer_span: mock.MagicMock,
+        _create_consumer_span: mock.AsyncMock,
         extract: mock.MagicMock,
     ) -> None:
         tracer = mock.MagicMock()
@@ -270,6 +270,7 @@ async def test_wrap_getmany(
             }
         )
         kafka_consumer = mock.MagicMock()
+        _create_consumer_span.return_value = mock.MagicMock()
 
         wrapped_getmany = _wrap_getmany(tracer, consume_hook)
         records = await wrapped_getmany(