fix(kafka): handle produce(key=None) [backport 1.20] (#7614)

Backport 68da170 from #7576 to 1.20.

This change fixes #7236 by falling back to the empty string when the
partition key passed to `produce()` is falsey.

## 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)
- [x] If this PR 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`.
- [x] This PR doesn't touch any of that.

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

Author: github-actions[bot]

ddtrace/contrib/kafka/patch.py

@@ -115,7 +115,7 @@ def traced_produce(func, instance, args, kwargs):
         value = get_argument_value(args, kwargs, 1, "value")
     except ArgumentError:
         value = None
-    message_key = kwargs.get("key", "")
+    message_key = kwargs.get("key", "") or ""
     partition = kwargs.get("partition", -1)
     core.dispatch("kafka.produce.start", [instance, args, kwargs])
 

releasenotes/notes/kafka-key-none-540d576e8590aa3c.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    kafka: This fix resolves an issue where calls to ``confluent_kafka``'s ``produce`` method with ``key=None``
+    would cause an exception to be raised.

tests/contrib/kafka/test_kafka.py

@@ -191,6 +191,15 @@ def test_produce_single_server(dummy_tracer, producer, kafka_topic):
     assert produce_span.get_tag("messaging.kafka.bootstrap.servers") == BOOTSTRAP_SERVERS
 
 
+def test_produce_none_key(dummy_tracer, producer, kafka_topic):
+    Pin.override(producer, tracer=dummy_tracer)
+    producer.produce(kafka_topic, PAYLOAD, key=None)
+    producer.flush()
+
+    traces = dummy_tracer.pop_traces()
+    assert 1 == len(traces), "key=None does not cause produce() call to raise an exception"
+
+
 def test_produce_multiple_servers(dummy_tracer, kafka_topic):
     producer = confluent_kafka.Producer({"bootstrap.servers": ",".join([BOOTSTRAP_SERVERS] * 3)})
     Pin.override(producer, tracer=dummy_tracer)