We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
AttributeError with kafka-python integration in 6.10.0
**Describe the bug**:

After upgrading the agent to 6.10.0, we immediately started seeing AttributeError failues in the ElasticAPM code when calling through to our kafka-python code paths (we've had to rollback to 6.9 because of this).

**To Reproduce**

Try to send a message using the KafkaProducer class in kafka-python.

Traceback:

```python
AttributeError: 'NoneType' object has no attribute 'id'
[...]
  File "elasticapm/instrumentation/packages/kafka.py", line 94, in call_if_sampling
    return self.call(module, method, wrapped, instance, args, kwargs)
  File "elasticapm/instrumentation/packages/kafka.py", line 106, in call
    return self._trace_send(instance, wrapped, destination_info=destination_info, *args, **kwargs)
  File "elasticapm/instrumentation/packages/kafka.py", line 71, in _trace_send
    tp = transaction.trace_parent.copy_from(span_id=span.id)
```

I have had to remove proprietary code from the stack trace. However, the two snipped lines of code before it hits the elasticapm code paths is basically just doing:

```python
from kafka import KafkaProducer
producer = KafkaProducer(**kafka_config)  # where kafka_config is a dict of config options.
producer.send(topic, message).get()
```

**Environment (please complete the following information)**
- OS: Linux
- Python version: 3.9.6-slim-bullseye (Docker container)
- Framework and version [e.g. Django 2.1]: Django 3.2.13, kafka-python 2.0.2
- Agent version: 6.10.0

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticapm/instrumentation/packages/kafka.py`
Content:
```
1 #  BSD 3-Clause License
2 #
3 #  Copyright (c) 2019, Elasticsearch BV
4 #  All rights reserved.
5 #
6 #  Redistribution and use in source and binary forms, with or without
7 #  modification, are permitted provided that the following conditions are met:
8 #
9 #  * Redistributions of source code must retain the above copyright notice, this
10 #    list of conditions and the following disclaimer.
11 #
12 #  * Redistributions in binary form must reproduce the above copyright notice,
13 #    this list of conditions and the following disclaimer in the documentation
14 #    and/or other materials provided with the distribution.
15 #
16 #  * Neither the name of the copyright holder nor the names of its
17 #    contributors may be used to endorse or promote products derived from
18 #    this software without specific prior written permission.
19 #
20 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
21 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
22 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
23 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
24 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
25 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
26 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
27 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
28 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
29 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
30 
31 import time
32 from typing import Optional
33 
34 import elasticapm
35 from elasticapm import get_client
36 from elasticapm.conf import constants
37 from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule
38 from elasticapm.traces import DroppedSpan, capture_span, execution_context
39 from elasticapm.utils.disttracing import TraceParent
40 
41 
42 class KafkaInstrumentation(AbstractInstrumentedModule):
43 
44     instrument_list = [
45         ("kafka", "KafkaProducer.send"),
46         ("kafka", "KafkaConsumer.poll"),
47         ("kafka", "KafkaConsumer.__next__"),
48     ]
49     provider_name = "kafka"
50     name = "kafka"
51 
52     def _trace_send(self, instance, wrapped, *args, destination_info=None, **kwargs):
53         topic = args[0] if args else kwargs["topic"]
54         headers = args[4] if len(args) > 4 else kwargs.get("headers", None)
55 
56         span_name = f"Kafka SEND to {topic}"
57         destination_info["service"]["resource"] += topic
58         with capture_span(
59             name=span_name,
60             span_type="messaging",
61             span_subtype=self.provider_name,
62             span_action="send",
63             leaf=True,
64             extra={
65                 "message": {"queue": {"name": topic}},
66                 "destination": destination_info,
67             },
68         ) as span:
69             transaction = execution_context.get_transaction()
70             if transaction:
71                 tp = transaction.trace_parent.copy_from(span_id=span.id)
72             if headers:
73                 headers.append((constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary()))
74             else:
75                 headers = [(constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary())]
76                 if len(args) > 4:
77                     args = list(args)
78                     args[4] = headers
79                 else:
80                     kwargs["headers"] = headers
81             result = wrapped(*args, **kwargs)
82             if instance and instance._metadata.controller and not isinstance(span, DroppedSpan):
83                 address = instance._metadata.controller[1]
84                 port = instance._metadata.controller[2]
85                 span.context["destination"]["address"] = address
86                 span.context["destination"]["port"] = port
87             return result
88 
89     def call_if_sampling(self, module, method, wrapped, instance, args, kwargs):
90         # Contrasting to the superclass implementation, we *always* want to
91         # return a proxied connection, even if there is no ongoing elasticapm
92         # transaction yet. This ensures that we instrument the cursor once
93         # the transaction started.
94         return self.call(module, method, wrapped, instance, args, kwargs)
95 
96     def call(self, module, method, wrapped, instance, args, kwargs):
97         client = get_client()
98         destination_info = {
99             "service": {"name": "kafka", "resource": "kafka/", "type": "messaging"},
100         }
101 
102         if method == "KafkaProducer.send":
103             topic = args[0] if args else kwargs["topic"]
104             if client.should_ignore_topic(topic) or not execution_context.get_transaction():
105                 return wrapped(*args, **kwargs)
106             return self._trace_send(instance, wrapped, destination_info=destination_info, *args, **kwargs)
107 
108         elif method == "KafkaConsumer.poll":
109             transaction = execution_context.get_transaction()
110             if transaction:
111                 with capture_span(
112                     name="Kafka POLL",
113                     span_type="messaging",
114                     span_subtype=self.provider_name,
115                     span_action="poll",
116                     leaf=True,
117                     extra={
118                         "destination": destination_info,
119                     },
120                 ) as span:
121                     if not isinstance(span, DroppedSpan) and instance._subscription.subscription:
122                         span.name += " from " + ", ".join(sorted(instance._subscription.subscription))
123                     results = wrapped(*args, **kwargs)
124                     return results
125             else:
126                 return wrapped(*args, **kwargs)
127 
128         elif method == "KafkaConsumer.__next__":
129             transaction = execution_context.get_transaction()
130             if transaction and transaction.transaction_type != "messaging":
131                 # somebody started a transaction outside of the consumer,
132                 # so we capture it as a span, and record the causal trace as a link
133                 with capture_span(
134                     name="consumer",
135                     span_type="messaging",
136                     span_subtype=self.provider_name,
137                     span_action="receive",
138                     leaf=True,
139                     extra={
140                         "message": {"queue": {"name": ""}},
141                         "destination": destination_info,
142                     },
143                 ) as span:
144                     try:
145                         result = wrapped(*args, **kwargs)
146                     except StopIteration:
147                         span.cancel()
148                         raise
149                     if not isinstance(span, DroppedSpan):
150                         topic = result[0]
151                         if client.should_ignore_topic(topic):
152                             span.cancel()
153                             return result
154                         trace_parent = self.get_traceparent_from_result(result)
155                         if trace_parent:
156                             span.add_link(trace_parent)
157                         destination_info["service"]["resource"] += topic
158                         span.context["message"]["queue"]["name"] = topic
159                         span.name = "Kafka RECEIVE from " + topic
160                     return result
161             else:
162                 # No transaction running, or this is a transaction started by us,
163                 # so let's end it and start the next,
164                 # unless a StopIteration is raised, at which point we do nothing.
165                 if transaction:
166                     client.end_transaction()
167                 result = wrapped(*args, **kwargs)
168                 topic = result[0]
169                 if client.should_ignore_topic(topic):
170                     return result
171                 trace_parent = self.get_traceparent_from_result(result)
172                 transaction = client.begin_transaction("messaging", trace_parent=trace_parent)
173                 if result.timestamp_type == 0:
174                     current_time_millis = int(round(time.time() * 1000))
175                     age = current_time_millis - result.timestamp
176                     transaction.context = {
177                         "message": {"age": {"ms": age}, "queue": {"name": topic}},
178                         "service": {"framework": {"name": "Kafka"}},
179                     }
180                 transaction_name = "Kafka RECEIVE from " + topic
181                 elasticapm.set_transaction_name(transaction_name, override=True)
182                 res = constants.OUTCOME.SUCCESS
183                 elasticapm.set_transaction_result(res, override=False)
184                 return result
185 
186     def get_traceparent_from_result(self, result) -> Optional[TraceParent]:
187         for k, v in result.headers:
188             if k == constants.TRACEPARENT_BINARY_HEADER_NAME:
189                 return TraceParent.from_binary(v)
190
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/elasticapm/instrumentation/packages/kafka.py b/elasticapm/instrumentation/packages/kafka.py
--- a/elasticapm/instrumentation/packages/kafka.py
+++ b/elasticapm/instrumentation/packages/kafka.py
@@ -36,7 +36,7 @@
 from elasticapm.conf import constants
 from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule
 from elasticapm.traces import DroppedSpan, capture_span, execution_context
-from elasticapm.utils.disttracing import TraceParent
+from elasticapm.utils.disttracing import TraceParent, TracingOptions
 
 
 class KafkaInstrumentation(AbstractInstrumentedModule):
@@ -48,6 +48,7 @@
     ]
     provider_name = "kafka"
     name = "kafka"
+    creates_transactions = True
 
     def _trace_send(self, instance, wrapped, *args, destination_info=None, **kwargs):
         topic = args[0] if args else kwargs["topic"]
@@ -68,7 +69,10 @@
         ) as span:
             transaction = execution_context.get_transaction()
             if transaction:
-                tp = transaction.trace_parent.copy_from(span_id=span.id)
+                tp = transaction.trace_parent.copy_from(
+                    span_id=span.id if span else transaction.id,
+                    trace_options=None if span else TracingOptions(recorded=False),
+                )
             if headers:
                 headers.append((constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary()))
             else:
@@ -79,22 +83,17 @@
                 else:
                     kwargs["headers"] = headers
             result = wrapped(*args, **kwargs)
-            if instance and instance._metadata.controller and not isinstance(span, DroppedSpan):
+            if span and instance and instance._metadata.controller and not isinstance(span, DroppedSpan):
                 address = instance._metadata.controller[1]
                 port = instance._metadata.controller[2]
                 span.context["destination"]["address"] = address
                 span.context["destination"]["port"] = port
             return result
 
-    def call_if_sampling(self, module, method, wrapped, instance, args, kwargs):
-        # Contrasting to the superclass implementation, we *always* want to
-        # return a proxied connection, even if there is no ongoing elasticapm
-        # transaction yet. This ensures that we instrument the cursor once
-        # the transaction started.
-        return self.call(module, method, wrapped, instance, args, kwargs)
-
     def call(self, module, method, wrapped, instance, args, kwargs):
         client = get_client()
+        if client is None:
+            return wrapped(*args, **kwargs)
         destination_info = {
             "service": {"name": "kafka", "resource": "kafka/", "type": "messaging"},
         }
@@ -118,7 +117,7 @@
                         "destination": destination_info,
                     },
                 ) as span:
-                    if not isinstance(span, DroppedSpan) and instance._subscription.subscription:
+                    if span and not isinstance(span, DroppedSpan) and instance._subscription.subscription:
                         span.name += " from " + ", ".join(sorted(instance._subscription.subscription))
                     results = wrapped(*args, **kwargs)
                     return results
@@ -146,7 +145,7 @@
                     except StopIteration:
                         span.cancel()
                         raise
-                    if not isinstance(span, DroppedSpan):
+                    if span and not isinstance(span, DroppedSpan):
                         topic = result[0]
                         if client.should_ignore_topic(topic):
                             span.cancel()

{"golden_diff": "diff --git a/elasticapm/instrumentation/packages/kafka.py b/elasticapm/instrumentation/packages/kafka.py\n--- a/elasticapm/instrumentation/packages/kafka.py\n+++ b/elasticapm/instrumentation/packages/kafka.py\n@@ -36,7 +36,7 @@\n from elasticapm.conf import constants\n from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\n from elasticapm.traces import DroppedSpan, capture_span, execution_context\n-from elasticapm.utils.disttracing import TraceParent\n+from elasticapm.utils.disttracing import TraceParent, TracingOptions\n \n \n class KafkaInstrumentation(AbstractInstrumentedModule):\n@@ -48,6 +48,7 @@\n     ]\n     provider_name = \"kafka\"\n     name = \"kafka\"\n+    creates_transactions = True\n \n     def _trace_send(self, instance, wrapped, *args, destination_info=None, **kwargs):\n         topic = args[0] if args else kwargs[\"topic\"]\n@@ -68,7 +69,10 @@\n         ) as span:\n             transaction = execution_context.get_transaction()\n             if transaction:\n-                tp = transaction.trace_parent.copy_from(span_id=span.id)\n+                tp = transaction.trace_parent.copy_from(\n+                    span_id=span.id if span else transaction.id,\n+                    trace_options=None if span else TracingOptions(recorded=False),\n+                )\n             if headers:\n                 headers.append((constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary()))\n             else:\n@@ -79,22 +83,17 @@\n                 else:\n                     kwargs[\"headers\"] = headers\n             result = wrapped(*args, **kwargs)\n-            if instance and instance._metadata.controller and not isinstance(span, DroppedSpan):\n+            if span and instance and instance._metadata.controller and not isinstance(span, DroppedSpan):\n                 address = instance._metadata.controller[1]\n                 port = instance._metadata.controller[2]\n                 span.context[\"destination\"][\"address\"] = address\n                 span.context[\"destination\"][\"port\"] = port\n             return result\n \n-    def call_if_sampling(self, module, method, wrapped, instance, args, kwargs):\n-        # Contrasting to the superclass implementation, we *always* want to\n-        # return a proxied connection, even if there is no ongoing elasticapm\n-        # transaction yet. This ensures that we instrument the cursor once\n-        # the transaction started.\n-        return self.call(module, method, wrapped, instance, args, kwargs)\n-\n     def call(self, module, method, wrapped, instance, args, kwargs):\n         client = get_client()\n+        if client is None:\n+            return wrapped(*args, **kwargs)\n         destination_info = {\n             \"service\": {\"name\": \"kafka\", \"resource\": \"kafka/\", \"type\": \"messaging\"},\n         }\n@@ -118,7 +117,7 @@\n                         \"destination\": destination_info,\n                     },\n                 ) as span:\n-                    if not isinstance(span, DroppedSpan) and instance._subscription.subscription:\n+                    if span and not isinstance(span, DroppedSpan) and instance._subscription.subscription:\n                         span.name += \" from \" + \", \".join(sorted(instance._subscription.subscription))\n                     results = wrapped(*args, **kwargs)\n                     return results\n@@ -146,7 +145,7 @@\n                     except StopIteration:\n                         span.cancel()\n                         raise\n-                    if not isinstance(span, DroppedSpan):\n+                    if span and not isinstance(span, DroppedSpan):\n                         topic = result[0]\n                         if client.should_ignore_topic(topic):\n                             span.cancel()\n", "issue": "AttributeError with kafka-python integration in 6.10.0\n**Describe the bug**:\r\n\r\nAfter upgrading the agent to 6.10.0, we immediately started seeing AttributeError failues in the ElasticAPM code when calling through to our kafka-python code paths (we've had to rollback to 6.9 because of this).\r\n\r\n**To Reproduce**\r\n\r\nTry to send a message using the KafkaProducer class in kafka-python.\r\n\r\nTraceback:\r\n\r\n```python\r\nAttributeError: 'NoneType' object has no attribute 'id'\r\n[...]\r\n  File \"elasticapm/instrumentation/packages/kafka.py\", line 94, in call_if_sampling\r\n    return self.call(module, method, wrapped, instance, args, kwargs)\r\n  File \"elasticapm/instrumentation/packages/kafka.py\", line 106, in call\r\n    return self._trace_send(instance, wrapped, destination_info=destination_info, *args, **kwargs)\r\n  File \"elasticapm/instrumentation/packages/kafka.py\", line 71, in _trace_send\r\n    tp = transaction.trace_parent.copy_from(span_id=span.id)\r\n```\r\n\r\nI have had to remove proprietary code from the stack trace. However, the two snipped lines of code before it hits the elasticapm code paths is basically just doing:\r\n\r\n```python\r\nfrom kafka import KafkaProducer\r\nproducer = KafkaProducer(**kafka_config)  # where kafka_config is a dict of config options.\r\nproducer.send(topic, message).get()\r\n```\r\n\r\n**Environment (please complete the following information)**\r\n- OS: Linux\r\n- Python version: 3.9.6-slim-bullseye (Docker container)\r\n- Framework and version [e.g. Django 2.1]: Django 3.2.13, kafka-python 2.0.2\r\n- Agent version: 6.10.0\n", "before_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nimport time\nfrom typing import Optional\n\nimport elasticapm\nfrom elasticapm import get_client\nfrom elasticapm.conf import constants\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import DroppedSpan, capture_span, execution_context\nfrom elasticapm.utils.disttracing import TraceParent\n\n\nclass KafkaInstrumentation(AbstractInstrumentedModule):\n\n    instrument_list = [\n        (\"kafka\", \"KafkaProducer.send\"),\n        (\"kafka\", \"KafkaConsumer.poll\"),\n        (\"kafka\", \"KafkaConsumer.__next__\"),\n    ]\n    provider_name = \"kafka\"\n    name = \"kafka\"\n\n    def _trace_send(self, instance, wrapped, *args, destination_info=None, **kwargs):\n        topic = args[0] if args else kwargs[\"topic\"]\n        headers = args[4] if len(args) > 4 else kwargs.get(\"headers\", None)\n\n        span_name = f\"Kafka SEND to {topic}\"\n        destination_info[\"service\"][\"resource\"] += topic\n        with capture_span(\n            name=span_name,\n            span_type=\"messaging\",\n            span_subtype=self.provider_name,\n            span_action=\"send\",\n            leaf=True,\n            extra={\n                \"message\": {\"queue\": {\"name\": topic}},\n                \"destination\": destination_info,\n            },\n        ) as span:\n            transaction = execution_context.get_transaction()\n            if transaction:\n                tp = transaction.trace_parent.copy_from(span_id=span.id)\n            if headers:\n                headers.append((constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary()))\n            else:\n                headers = [(constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary())]\n                if len(args) > 4:\n                    args = list(args)\n                    args[4] = headers\n                else:\n                    kwargs[\"headers\"] = headers\n            result = wrapped(*args, **kwargs)\n            if instance and instance._metadata.controller and not isinstance(span, DroppedSpan):\n                address = instance._metadata.controller[1]\n                port = instance._metadata.controller[2]\n                span.context[\"destination\"][\"address\"] = address\n                span.context[\"destination\"][\"port\"] = port\n            return result\n\n    def call_if_sampling(self, module, method, wrapped, instance, args, kwargs):\n        # Contrasting to the superclass implementation, we *always* want to\n        # return a proxied connection, even if there is no ongoing elasticapm\n        # transaction yet. This ensures that we instrument the cursor once\n        # the transaction started.\n        return self.call(module, method, wrapped, instance, args, kwargs)\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        client = get_client()\n        destination_info = {\n            \"service\": {\"name\": \"kafka\", \"resource\": \"kafka/\", \"type\": \"messaging\"},\n        }\n\n        if method == \"KafkaProducer.send\":\n            topic = args[0] if args else kwargs[\"topic\"]\n            if client.should_ignore_topic(topic) or not execution_context.get_transaction():\n                return wrapped(*args, **kwargs)\n            return self._trace_send(instance, wrapped, destination_info=destination_info, *args, **kwargs)\n\n        elif method == \"KafkaConsumer.poll\":\n            transaction = execution_context.get_transaction()\n            if transaction:\n                with capture_span(\n                    name=\"Kafka POLL\",\n                    span_type=\"messaging\",\n                    span_subtype=self.provider_name,\n                    span_action=\"poll\",\n                    leaf=True,\n                    extra={\n                        \"destination\": destination_info,\n                    },\n                ) as span:\n                    if not isinstance(span, DroppedSpan) and instance._subscription.subscription:\n                        span.name += \" from \" + \", \".join(sorted(instance._subscription.subscription))\n                    results = wrapped(*args, **kwargs)\n                    return results\n            else:\n                return wrapped(*args, **kwargs)\n\n        elif method == \"KafkaConsumer.__next__\":\n            transaction = execution_context.get_transaction()\n            if transaction and transaction.transaction_type != \"messaging\":\n                # somebody started a transaction outside of the consumer,\n                # so we capture it as a span, and record the causal trace as a link\n                with capture_span(\n                    name=\"consumer\",\n                    span_type=\"messaging\",\n                    span_subtype=self.provider_name,\n                    span_action=\"receive\",\n                    leaf=True,\n                    extra={\n                        \"message\": {\"queue\": {\"name\": \"\"}},\n                        \"destination\": destination_info,\n                    },\n                ) as span:\n                    try:\n                        result = wrapped(*args, **kwargs)\n                    except StopIteration:\n                        span.cancel()\n                        raise\n                    if not isinstance(span, DroppedSpan):\n                        topic = result[0]\n                        if client.should_ignore_topic(topic):\n                            span.cancel()\n                            return result\n                        trace_parent = self.get_traceparent_from_result(result)\n                        if trace_parent:\n                            span.add_link(trace_parent)\n                        destination_info[\"service\"][\"resource\"] += topic\n                        span.context[\"message\"][\"queue\"][\"name\"] = topic\n                        span.name = \"Kafka RECEIVE from \" + topic\n                    return result\n            else:\n                # No transaction running, or this is a transaction started by us,\n                # so let's end it and start the next,\n                # unless a StopIteration is raised, at which point we do nothing.\n                if transaction:\n                    client.end_transaction()\n                result = wrapped(*args, **kwargs)\n                topic = result[0]\n                if client.should_ignore_topic(topic):\n                    return result\n                trace_parent = self.get_traceparent_from_result(result)\n                transaction = client.begin_transaction(\"messaging\", trace_parent=trace_parent)\n                if result.timestamp_type == 0:\n                    current_time_millis = int(round(time.time() * 1000))\n                    age = current_time_millis - result.timestamp\n                    transaction.context = {\n                        \"message\": {\"age\": {\"ms\": age}, \"queue\": {\"name\": topic}},\n                        \"service\": {\"framework\": {\"name\": \"Kafka\"}},\n                    }\n                transaction_name = \"Kafka RECEIVE from \" + topic\n                elasticapm.set_transaction_name(transaction_name, override=True)\n                res = constants.OUTCOME.SUCCESS\n                elasticapm.set_transaction_result(res, override=False)\n                return result\n\n    def get_traceparent_from_result(self, result) -> Optional[TraceParent]:\n        for k, v in result.headers:\n            if k == constants.TRACEPARENT_BINARY_HEADER_NAME:\n                return TraceParent.from_binary(v)\n", "path": "elasticapm/instrumentation/packages/kafka.py"}], "after_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nimport time\nfrom typing import Optional\n\nimport elasticapm\nfrom elasticapm import get_client\nfrom elasticapm.conf import constants\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import DroppedSpan, capture_span, execution_context\nfrom elasticapm.utils.disttracing import TraceParent, TracingOptions\n\n\nclass KafkaInstrumentation(AbstractInstrumentedModule):\n\n    instrument_list = [\n        (\"kafka\", \"KafkaProducer.send\"),\n        (\"kafka\", \"KafkaConsumer.poll\"),\n        (\"kafka\", \"KafkaConsumer.__next__\"),\n    ]\n    provider_name = \"kafka\"\n    name = \"kafka\"\n    creates_transactions = True\n\n    def _trace_send(self, instance, wrapped, *args, destination_info=None, **kwargs):\n        topic = args[0] if args else kwargs[\"topic\"]\n        headers = args[4] if len(args) > 4 else kwargs.get(\"headers\", None)\n\n        span_name = f\"Kafka SEND to {topic}\"\n        destination_info[\"service\"][\"resource\"] += topic\n        with capture_span(\n            name=span_name,\n            span_type=\"messaging\",\n            span_subtype=self.provider_name,\n            span_action=\"send\",\n            leaf=True,\n            extra={\n                \"message\": {\"queue\": {\"name\": topic}},\n                \"destination\": destination_info,\n            },\n        ) as span:\n            transaction = execution_context.get_transaction()\n            if transaction:\n                tp = transaction.trace_parent.copy_from(\n                    span_id=span.id if span else transaction.id,\n                    trace_options=None if span else TracingOptions(recorded=False),\n                )\n            if headers:\n                headers.append((constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary()))\n            else:\n                headers = [(constants.TRACEPARENT_BINARY_HEADER_NAME, tp.to_binary())]\n                if len(args) > 4:\n                    args = list(args)\n                    args[4] = headers\n                else:\n                    kwargs[\"headers\"] = headers\n            result = wrapped(*args, **kwargs)\n            if span and instance and instance._metadata.controller and not isinstance(span, DroppedSpan):\n                address = instance._metadata.controller[1]\n                port = instance._metadata.controller[2]\n                span.context[\"destination\"][\"address\"] = address\n                span.context[\"destination\"][\"port\"] = port\n            return result\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        client = get_client()\n        if client is None:\n            return wrapped(*args, **kwargs)\n        destination_info = {\n            \"service\": {\"name\": \"kafka\", \"resource\": \"kafka/\", \"type\": \"messaging\"},\n        }\n\n        if method == \"KafkaProducer.send\":\n            topic = args[0] if args else kwargs[\"topic\"]\n            if client.should_ignore_topic(topic) or not execution_context.get_transaction():\n                return wrapped(*args, **kwargs)\n            return self._trace_send(instance, wrapped, destination_info=destination_info, *args, **kwargs)\n\n        elif method == \"KafkaConsumer.poll\":\n            transaction = execution_context.get_transaction()\n            if transaction:\n                with capture_span(\n                    name=\"Kafka POLL\",\n                    span_type=\"messaging\",\n                    span_subtype=self.provider_name,\n                    span_action=\"poll\",\n                    leaf=True,\n                    extra={\n                        \"destination\": destination_info,\n                    },\n                ) as span:\n                    if span and not isinstance(span, DroppedSpan) and instance._subscription.subscription:\n                        span.name += \" from \" + \", \".join(sorted(instance._subscription.subscription))\n                    results = wrapped(*args, **kwargs)\n                    return results\n            else:\n                return wrapped(*args, **kwargs)\n\n        elif method == \"KafkaConsumer.__next__\":\n            transaction = execution_context.get_transaction()\n            if transaction and transaction.transaction_type != \"messaging\":\n                # somebody started a transaction outside of the consumer,\n                # so we capture it as a span, and record the causal trace as a link\n                with capture_span(\n                    name=\"consumer\",\n                    span_type=\"messaging\",\n                    span_subtype=self.provider_name,\n                    span_action=\"receive\",\n                    leaf=True,\n                    extra={\n                        \"message\": {\"queue\": {\"name\": \"\"}},\n                        \"destination\": destination_info,\n                    },\n                ) as span:\n                    try:\n                        result = wrapped(*args, **kwargs)\n                    except StopIteration:\n                        span.cancel()\n                        raise\n                    if span and not isinstance(span, DroppedSpan):\n                        topic = result[0]\n                        if client.should_ignore_topic(topic):\n                            span.cancel()\n                            return result\n                        trace_parent = self.get_traceparent_from_result(result)\n                        if trace_parent:\n                            span.add_link(trace_parent)\n                        destination_info[\"service\"][\"resource\"] += topic\n                        span.context[\"message\"][\"queue\"][\"name\"] = topic\n                        span.name = \"Kafka RECEIVE from \" + topic\n                    return result\n            else:\n                # No transaction running, or this is a transaction started by us,\n                # so let's end it and start the next,\n                # unless a StopIteration is raised, at which point we do nothing.\n                if transaction:\n                    client.end_transaction()\n                result = wrapped(*args, **kwargs)\n                topic = result[0]\n                if client.should_ignore_topic(topic):\n                    return result\n                trace_parent = self.get_traceparent_from_result(result)\n                transaction = client.begin_transaction(\"messaging\", trace_parent=trace_parent)\n                if result.timestamp_type == 0:\n                    current_time_millis = int(round(time.time() * 1000))\n                    age = current_time_millis - result.timestamp\n                    transaction.context = {\n                        \"message\": {\"age\": {\"ms\": age}, \"queue\": {\"name\": topic}},\n                        \"service\": {\"framework\": {\"name\": \"Kafka\"}},\n                    }\n                transaction_name = \"Kafka RECEIVE from \" + topic\n                elasticapm.set_transaction_name(transaction_name, override=True)\n                res = constants.OUTCOME.SUCCESS\n                elasticapm.set_transaction_result(res, override=False)\n                return result\n\n    def get_traceparent_from_result(self, result) -> Optional[TraceParent]:\n        for k, v in result.headers:\n            if k == constants.TRACEPARENT_BINARY_HEADER_NAME:\n                return TraceParent.from_binary(v)\n", "path": "elasticapm/instrumentation/packages/kafka.py"}]}