fix: IndexError in the kafka_python.py

Fixed potential IndexError in `kafka_python.py` by handling both positional and keyword arguments for the topic parameter in the `trace_kafka_send` function. The issue is similar to the one fixed in `confluent_kafka_python.py`, where an IndexError could occur when the topic was passed as a keyword argument, resulting in an empty `args` tuple.

The solution:
1. Modified the `trace_kafka_send` function to get the topic from either `args` or `kwargs`
2. Added safety checks to handle edge cases
3. Added two new test methods to verify the fix works with different argument patterns:
   - `test_trace_kafka_python_send_with_keyword_topic`
   - `test_trace_kafka_python_send_with_keyword_args`

This fix ensures that the Kafka instrumentation works correctly regardless of how the `send` method is called, improving the robustness of the Python sensor.

Signed-off-by: Paulo Vital <[email protected]>

Author: pvital

src/instana/instrumentation/kafka/kafka_python.py

@@ -14,11 +14,8 @@
     from instana.log import logger
     from instana.propagators.format import Format
     from instana.singletons import get_tracer
-    from instana.util.traceutils import (
-        get_tracer_tuple,
-        tracing_is_off,
-    )
     from instana.span.span import InstanaSpan
+    from instana.util.traceutils import get_tracer_tuple, tracing_is_off
 
     if TYPE_CHECKING:
         from kafka.producer.future import FutureRecordMetadata
@@ -38,15 +35,25 @@ def trace_kafka_send(
 
         tracer, parent_span, _ = get_tracer_tuple()
         parent_context = parent_span.get_span_context() if parent_span else None
+
+        # Get the topic from either args or kwargs
+        topic = args[0] if args else kwargs.get("topic")
+
+        # Handle the case where topic might not be available
+        if topic is None:
+            # Log a warning and proceed with the original function call
+            logger.warning("Kafka send called without a topic parameter")
+            return wrapped(*args, **kwargs)
+
         is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
             "kafka",
             "send",
-            args[0],
+            topic,
         )
         with tracer.start_as_current_span(
             "kafka-producer", span_context=parent_context, kind=SpanKind.PRODUCER
         ) as span:
-            span.set_attribute("kafka.service", args[0])
+            span.set_attribute("kafka.service", topic)
             span.set_attribute("kafka.access", "send")
 
             # context propagation

tests/clients/kafka/test_kafka_python.py

@@ -12,19 +12,18 @@
 from opentelemetry.trace.span import format_span_id
 
 from instana.configurator import config
-from instana.options import StandardOptions
-from instana.singletons import agent, tracer
-from instana.util.config import parse_ignored_endpoints_from_yaml
-from tests.helpers import get_first_span_by_filter, testenv
-
 from instana.instrumentation.kafka import kafka_python
 from instana.instrumentation.kafka.kafka_python import (
     clear_context,
-    save_consumer_span_into_context,
     close_consumer_span,
     consumer_span,
+    save_consumer_span_into_context,
 )
+from instana.options import StandardOptions
+from instana.singletons import agent, tracer
 from instana.span.span import InstanaSpan
+from instana.util.config import parse_ignored_endpoints_from_yaml
+from tests.helpers import get_first_span_by_filter, testenv
 
 
 class TestKafkaPython:
@@ -122,6 +121,70 @@ def test_trace_kafka_python_send(self) -> None:
         assert kafka_span.data["kafka"]["service"] == testenv["kafka_topic"]
         assert kafka_span.data["kafka"]["access"] == "send"
 
+    def test_trace_kafka_python_send_with_keyword_topic(self) -> None:
+        """Test that tracing works when topic is passed as a keyword argument."""
+        with tracer.start_as_current_span("test"):
+            # Pass topic as a keyword argument
+            future = self.producer.send(
+                topic=testenv["kafka_topic"], value=b"raw_bytes"
+            )
+
+        _ = future.get(timeout=10)  # noqa: F841
+
+        spans = self.recorder.queued_spans()
+        assert len(spans) == 2
+
+        kafka_span = spans[0]
+        test_span = spans[1]
+
+        # Same traceId
+        assert test_span.t == kafka_span.t
+
+        # Parent relationships
+        assert kafka_span.p == test_span.s
+
+        # Error logging
+        assert not test_span.ec
+        assert not kafka_span.ec
+
+        assert kafka_span.n == "kafka"
+        assert kafka_span.k == SpanKind.CLIENT
+        assert kafka_span.data["kafka"]["service"] == testenv["kafka_topic"]
+        assert kafka_span.data["kafka"]["access"] == "send"
+
+    def test_trace_kafka_python_send_with_keyword_args(self) -> None:
+        """Test that tracing works when both topic and headers are passed as keyword arguments."""
+        with tracer.start_as_current_span("test"):
+            # Pass both topic and headers as keyword arguments
+            future = self.producer.send(
+                topic=testenv["kafka_topic"],
+                value=b"raw_bytes",
+                headers=[("custom-header", b"header-value")],
+            )
+
+        _ = future.get(timeout=10)  # noqa: F841
+
+        spans = self.recorder.queued_spans()
+        assert len(spans) == 2
+
+        kafka_span = spans[0]
+        test_span = spans[1]
+
+        # Same traceId
+        assert test_span.t == kafka_span.t
+
+        # Parent relationships
+        assert kafka_span.p == test_span.s
+
+        # Error logging
+        assert not test_span.ec
+        assert not kafka_span.ec
+
+        assert kafka_span.n == "kafka"
+        assert kafka_span.k == SpanKind.CLIENT
+        assert kafka_span.data["kafka"]["service"] == testenv["kafka_topic"]
+        assert kafka_span.data["kafka"]["access"] == "send"
+
     def test_trace_kafka_python_consume(self) -> None:
         # Produce some events
         self.producer.send(testenv["kafka_topic"], b"raw_bytes1")