wip: added context propagation to the kafka spans

Signed-off-by: Cagri Yonca <cagri@ibm.com>

Author: CagriYonca

src/instana/agent/host.py

@@ -360,6 +360,12 @@ def filter_spans(self, spans: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
                 )
                 if service == "kafka":
                     endpoint = span.data[service][service_specifier_key]
+                # comment these out only if you like to ignore internal events
+                # if (
+                #     service == "urllib3"
+                #     and "com.instana" in span.data["http"]["url"]
+                # ):
+                #     continue
                 method = span.data[service][operation_specifier_key]
                 if isinstance(method, str) and self.__is_endpoint_ignored(
                     service, method, endpoint

src/instana/instrumentation/kafka/kafka_python.py

@@ -37,14 +37,16 @@ def trace_kafka_send(
             span.set_attribute("kafka.access", "send")
 
             # context propagation
+            headers = kwargs.get("headers", [])
             tracer.inject(
                 span.context,
                 Format.KAFKA_HEADERS,
-                kwargs.get("headers", {}),
+                headers,
                 disable_w3c_trace_context=True,
             )
 
             try:
+                kwargs["headers"] = headers
                 res = wrapped(*args, **kwargs)
             except Exception as exc:
                 span.record_exception(exc)
@@ -62,28 +64,30 @@ def trace_kafka_consume(
             return wrapped(*args, **kwargs)
 
         tracer, parent_span, _ = get_tracer_tuple()
-
-        parent_context = (
-            parent_span.get_span_context()
-            if parent_span
-            else tracer.extract(
-                Format.KAFKA_HEADERS, {}, disable_w3c_trace_context=True
+        exception = ""
+        try:
+            res = wrapped(*args, **kwargs)
+        except Exception as exc:
+            exception = exc
+        else:
+            parent_context = (
+                parent_span.get_span_context()
+                if parent_span
+                else tracer.extract(
+                    Format.KAFKA_HEADERS,
+                    res.headers,
+                    disable_w3c_trace_context=True,
+                )
             )
-        )
-
-        with tracer.start_as_current_span(
-            "kafka-consumer", span_context=parent_context, kind=SpanKind.CONSUMER
-        ) as span:
-            topic = list(instance.subscription())[0]
-            span.set_attribute("kafka.service", topic)
-            span.set_attribute("kafka.access", "consume")
-
-            try:
-                res = wrapped(*args, **kwargs)
-            except Exception as exc:
-                span.record_exception(exc)
-            else:
-                return res
+            with tracer.start_as_current_span(
+                "kafka-consumer", span_context=parent_context, kind=SpanKind.CONSUMER
+            ) as span:
+                span.set_attribute("kafka.service", res.topic)
+                span.set_attribute("kafka.access", "consume")
+
+                if exception:
+                    span.record_exception(exception)
+        return res
 
     @wrapt.patch_function_wrapper("kafka", "KafkaConsumer.poll")
     def trace_kafka_poll(