fix: IndexError in the confluent_kafka_python.py

Fixed IndexError in `confluent_kafka_python.py` by handling both positional and keyword arguments for the topic parameter in the `trace_kafka_produce` function. The issue occurred when the topic was passed as a keyword argument, resulting in an empty args tuple and causing an IndexError when trying to access `args[0]`.

The solution:
1. Modified the `trace_kafka_produce` 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_confluent_kafka_produce_with_keyword_topic`
   - `test_trace_confluent_kafka_produce_with_keyword_args`

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

Author: pvital

src/instana/instrumentation/kafka/confluent_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
 
     consumer_token = None
     consumer_span = contextvars.ContextVar("confluent_kafka_consumer_span")
@@ -69,16 +66,26 @@ def trace_kafka_produce(
 
         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 produce called without a topic parameter")
+            return wrapped(*args, **kwargs)
+
         is_suppressed = tracer.exporter._HostAgent__is_endpoint_ignored(
             "kafka",
             "produce",
-            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", "produce")
 
             # context propagation
@@ -89,6 +96,10 @@ def trace_kafka_produce(
             # dictionary. To maintain compatibility with the headers for the
             # Kafka Python library, we will use a list of tuples.
             headers = args[6] if len(args) > 6 else kwargs.get("headers", [])
+
+            # Initialize headers if it's None
+            if headers is None:
+                headers = []
             suppression_header = {"x_instana_l_s": "0" if is_suppressed else "1"}
             headers.append(suppression_header)
 

tests/clients/kafka/test_confluent_kafka.py

@@ -5,30 +5,26 @@
 from typing import Generator
 
 import pytest
-from confluent_kafka import (
-    Consumer,
-    KafkaException,
-    Producer,
-)
+from confluent_kafka import Consumer, KafkaException, Producer
 from confluent_kafka.admin import AdminClient, NewTopic
-from mock import patch, Mock
+from mock import Mock, patch
 from opentelemetry.trace import SpanKind
 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 confluent_kafka_python
 from instana.instrumentation.kafka.confluent_kafka_python import (
     clear_context,
-    save_consumer_span_into_context,
     close_consumer_span,
-    trace_kafka_close,
     consumer_span,
+    save_consumer_span_into_context,
+    trace_kafka_close,
 )
+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 TestConfluentKafka:
@@ -120,6 +116,66 @@ def test_trace_confluent_kafka_produce(self) -> None:
         assert kafka_span.data["kafka"]["service"] == testenv["kafka_topic"]
         assert kafka_span.data["kafka"]["access"] == "produce"
 
+    def test_trace_confluent_kafka_produce_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
+            self.producer.produce(topic=testenv["kafka_topic"], value=b"raw_bytes")
+            self.producer.flush(timeout=10)
+
+        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"] == "produce"
+
+    def test_trace_confluent_kafka_produce_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
+            self.producer.produce(
+                topic=testenv["kafka_topic"],
+                value=b"raw_bytes",
+                headers=[("custom-header", b"header-value")],
+            )
+            self.producer.flush(timeout=10)
+
+        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"] == "produce"
+
     def test_trace_confluent_kafka_consume(self) -> None:
         agent.options.set_trace_configurations()
         # Produce some events