fix(kafka): handle errors from ensure_text gracefully for Kafka tracing [backport 1.17] (#6514)

Backport 8536722 from #6493 to 1.17.

## Motivation
This change ensures that we don't hard fail if there's string decoded by
`ensure_text` isn't utf-8 encoded. We saw this in one our production
applications after updating this library. While we are investigating the
messages to see if there's an issue in our producer, we still would
prefer that the code not hard error at this point if it's being traced.

```python

Traceback (most recent call last):
....
  File "/pyenv/versions/.../lib/python3.8/site-packages/confluent_kafka/avro/__init__.py", line 174, in poll
    message = super(AvroConsumer, self).poll(timeout)
  File "/pyenv/versions/.../lib/python3.8/site-packages/ddtrace/contrib/kafka/patch.py", line 157, in traced_poll
    span.set_tag_str(kafkax.MESSAGE_KEY, ensure_text(message_key))
  File "/pyenv/versions/.../lib/python3.8/site-packages/six.py", line 951, in ensure_text
    return s.decode(encoding, errors)
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xc8 in position 5: invalid continuation byte

```

This is a non-breaking change (can't seem to edit the labels of this
PR).

## 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] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [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))

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly 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: Joe Danis <[email protected]>

Author: github-actions[bot]

ddtrace/contrib/kafka/patch.py

@@ -116,7 +116,7 @@ def traced_produce(func, instance, args, kwargs):
         span.set_tag_str(COMPONENT, config.kafka.integration_name)
         span.set_tag_str(SPAN_KIND, SpanKind.PRODUCER)
         span.set_tag_str(kafkax.TOPIC, topic)
-        span.set_tag_str(kafkax.MESSAGE_KEY, ensure_text(message_key))
+        span.set_tag_str(kafkax.MESSAGE_KEY, ensure_text(message_key, errors="replace"))
         span.set_tag(kafkax.PARTITION, partition)
         span.set_tag_str(kafkax.TOMBSTONE, str(value is None))
         span.set_tag(SPAN_MEASURED_KEY)
@@ -154,7 +154,7 @@ def traced_poll(func, instance, args, kwargs):
             message_key = message.key() or ""
             message_offset = message.offset() or -1
             span.set_tag_str(kafkax.TOPIC, message.topic())
-            span.set_tag_str(kafkax.MESSAGE_KEY, ensure_text(message_key))
+            span.set_tag_str(kafkax.MESSAGE_KEY, ensure_text(message_key, errors="replace"))
             span.set_tag(kafkax.PARTITION, message.partition())
             span.set_tag_str(kafkax.TOMBSTONE, str(len(message) == 0))
             span.set_tag(kafkax.MESSAGE_OFFSET, message_offset)

releasenotes/notes/handle_ensure_text_errors_gracefully-63bcee9f2c3f3d96.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    kafka: Resolves ``UnicodeDecodeError`` raised when kafka messages key contain characters that are not supported by UTF-8 encoding.