diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_instrumentation.py
index 67ef3c2d0c..1885c278f1 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_instrumentation.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_instrumentation.py
@@ -1,12 +1,14 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
-from unittest import TestCase
+from unittest import TestCase, mock
 
 from kafka import KafkaConsumer, KafkaProducer
 from wrapt import BoundFunctionWrapper
 
+from opentelemetry import trace
 from opentelemetry.instrumentation.kafka import KafkaInstrumentor
 from opentelemetry.instrumentation.kafka.package import _instruments
+from opentelemetry.trace import NoOpTracerProvider
 
 
 class TestKafka(TestCase):
@@ -25,3 +27,43 @@ def test_instrumentation_dependencies(self) -> None:
         instrumentation = KafkaInstrumentor()
 
         self.assertEqual(instrumentation.instrumentation_dependencies(), _instruments)
+
+    def test_instrument_with_noop_tracer_provider(self) -> None:
+        instrumentation = KafkaInstrumentor()
+        provider = NoOpTracerProvider()
+        produce_hook = mock.Mock()
+
+        original_get_tracer = trace.get_tracer
+
+        with (
+            mock.patch.object(KafkaProducer, "send", autospec=True) as send_mock,
+            mock.patch(
+                "opentelemetry.instrumentation.kafka.trace.get_tracer",
+                wraps=original_get_tracer,
+            ) as get_tracer_mock,
+        ):
+            instrumentation.instrument(
+                tracer_provider=provider,
+                produce_hook=produce_hook,
+            )
+
+            try:
+                producer = object.__new__(KafkaProducer)
+                producer.config = {"bootstrap_servers": ["localhost:9092"]}
+
+                result = producer.send("test_topic", b"test_message")
+
+                send_mock.assert_called_once()
+                self.assertIs(result, send_mock.return_value)
+
+                self.assertIs(
+                    get_tracer_mock.call_args.kwargs["tracer_provider"],
+                    provider,
+                )
+
+                produce_hook.assert_called_once()
+                span = produce_hook.call_args.args[0]
+                self.assertFalse(span.is_recording())
+                self.assertFalse(span.get_span_context().is_valid)
+            finally:
+                instrumentation.uninstrument()
diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py
index 4c6e6a97dc..602a5886f1 100644
--- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py
+++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py
@@ -21,7 +21,7 @@
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.test.samplertestutil import CapturingSampler
 from opentelemetry.test.test_base import TestBase
-from opentelemetry.trace import SpanKind, StatusCode
+from opentelemetry.trace import NoOpTracerProvider, SpanKind, StatusCode
 
 
 class TestUtils(TestCase):
@@ -405,3 +405,52 @@ def test_consumer_error_propagates(self) -> None:
 
         self.assertIs(raised.exception, error)
         self.assertEqual(len(self.memory_exporter.get_finished_spans()), 0)
+
+
+class TestNoOpTracerProvider(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.tracer = NoOpTracerProvider().get_tracer(__name__)
+        self.config = {"bootstrap_servers": ["localhost:9092"]}
+
+    def test_noop_producer(self) -> None:
+        producer = mock.MagicMock()
+        producer.config = self.config
+
+        original_send = mock.Mock()
+        produce_hook = mock.Mock()
+
+        args = ["test_topic"]
+        kwargs = {"headers": []}
+
+        wrapped_send = _wrap_send(self.tracer, produce_hook)
+        result = wrapped_send(original_send, producer, args, kwargs)
+
+        original_send.assert_called_once_with(*args, **kwargs)
+        self.assertIs(result, original_send.return_value)
+
+        produce_hook.assert_called_once()
+        span = produce_hook.call_args.args[0]
+        self.assertFalse(span.is_recording())
+
+    def test_noop_consumer(self) -> None:
+        consumer = mock.MagicMock()
+        consumer.config = self.config
+
+        record = mock.Mock(
+            topic="test_topic",
+            partition=0,
+            headers=[],
+        )
+        original_next = mock.Mock(return_value=record)
+        consume_hook = mock.Mock()
+
+        wrapped_next = _wrap_next(self.tracer, consume_hook)
+        result = wrapped_next(original_next, consumer, [], {})
+
+        original_next.assert_called_once_with()
+        self.assertIs(result, record)
+
+        consume_hook.assert_called_once()
+        span = consume_hook.call_args.args[0]
+        self.assertFalse(span.is_recording())