Fix to_binary crash on missing datacontenttype or non-string attrs

The Kafka binary conversion to_binary() read the optional datacontenttype
attribute with event["datacontenttype"], raising KeyError for any event that
does not set it (datacontenttype is optional in the CloudEvents spec). It also
called .encode() directly on attribute values, raising AttributeError for
spec-legal non-string extension attributes (e.g. integers).

Read datacontenttype via .get() -- matching the sibling to_structured(), which
already guards this attribute -- and stringify attribute values before encoding,
matching the newer core Kafka binding. Add regression tests for both cases.

Signed-off-by: Md. Amdadul Bari Imad <amdadulbari@gmail.com>

Author: amdadulbari

src/cloudevents/v1/kafka/conversion.py

@@ -92,12 +92,13 @@ def to_binary(
         )
 
     headers = {}
-    if event["datacontenttype"]:
-        headers["content-type"] = event["datacontenttype"].encode("utf-8")
+    datacontenttype = event.get("datacontenttype")
+    if datacontenttype:
+        headers["content-type"] = datacontenttype.encode("utf-8")
     for attr, value in event.get_attributes().items():
         if attr not in ["data", "partitionkey", "datacontenttype"]:
             if value is not None:
-                headers["ce_{0}".format(attr)] = value.encode("utf-8")
+                headers["ce_{0}".format(attr)] = str(value).encode("utf-8")
 
     try:
         data = data_marshaller(event.get_data())

tests/test_v1_compat/test_kafka_conversions.py

@@ -129,6 +129,20 @@ def test_sets_headers(self, source_event):
         assert "data" not in result.headers
         assert "partitionkey" not in result.headers
 
+    def test_no_datacontenttype(self, source_event):
+        # datacontenttype is optional; to_binary must not raise KeyError when it
+        # is absent, and should simply omit the content-type header.
+        del source_event["datacontenttype"]
+        result = to_binary(source_event)
+        assert "content-type" not in result.headers
+
+    def test_non_string_extension_attribute(self, source_event):
+        # Extension attributes may be non-string (e.g. int) per the CloudEvents
+        # spec; to_binary must stringify them rather than raising AttributeError.
+        source_event["extension1"] = 5
+        result = to_binary(source_event)
+        assert result.headers["ce_extension1"] == b"5"
+
     def test_raise_marshaller_exception(self, source_event):
         with pytest.raises(cloud_exceptions.DataMarshallerError):
             to_binary(source_event, data_marshaller=failing_func)