fix(kafka): catch typeerror during tombstone tagging [backport 2.5] (#8508)

Backport cf942b4 from #8506 to 2.5.

This pull request resolves
#8016 by catching an error
that can occur during message consumption due to
confluentinc/confluent-kafka-python#1192.

## Checklist

- [x] Change(s) are motivated and described in the PR description
- [x] Testing strategy is described if automated tests are not included
in the PR
- [x] Risks are described (performance impact, potential for breakage,
maintainability)
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed or label `changelog/no-changelog` is set
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/))
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))
- [x] If this PR changes the public interface, I've notified
`@DataDog/apm-tees`.
- [x] If change touches code that signs or publishes builds or packages,
or handles credentials of any kind, I've requested a review from
`@DataDog/security-design-and-guidance`.

## Reviewer Checklist

- [x] Title is accurate
- [x] All changes are related to the pull request's stated goal
- [x] Description motivates each change
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- [x] Testing strategy adequately addresses listed risks
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] Release note makes sense to a user of the library
- [x] Author has acknowledged and discussed the performance implications
of this PR as reported in the benchmarks PR comment
- [x] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Emmett Butler <[email protected]>
Co-authored-by: Federico Mon <[email protected]>

Author: 3 people

ddtrace/contrib/kafka/patch.py

@@ -244,7 +244,12 @@ def traced_poll(func, instance, args, kwargs):
                 ):
                     span.set_tag_str(kafkax.MESSAGE_KEY, message_key)
                 span.set_tag(kafkax.PARTITION, message.partition())
-                span.set_tag_str(kafkax.TOMBSTONE, str(len(message) == 0))
+                is_tombstone = False
+                try:
+                    is_tombstone = len(message) == 0
+                except TypeError:  # https://github.com/confluentinc/confluent-kafka-python/issues/1192
+                    pass
+                span.set_tag_str(kafkax.TOMBSTONE, str(is_tombstone))
                 span.set_tag(kafkax.MESSAGE_OFFSET, message_offset)
             span.set_tag(SPAN_MEASURED_KEY)
             rate = config.kafka.get_analytics_sample_rate()

releasenotes/notes/kafka-typeerror-b39a35c5338b05aa.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    kafka: This fix resolves an issue where the use of a Kafka ``DeserializingConsumer`` could result in
+    a crash when the deserializer in use returns a type without a ``__len__`` attribute.

tests/contrib/kafka/test_kafka.py

@@ -833,7 +833,7 @@ def json_deserializer(as_bytes, ctx):
         try:
             return json.loads(as_bytes)
         except json.decoder.JSONDecodeError:
-            return as_bytes
+            return  # return a type that has no __len__ because such types caused a crash at one point
 
     conf = {
         "bootstrap.servers": BOOTSTRAP_SERVERS,