diff --git a/CHANGELOG.md b/CHANGELOG.md
index aa654e2a..af32d4de 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed `kafka.to_binary()` raising `KeyError` for events without a
+  `datacontenttype` and `AttributeError` for non-string attribute values; it now
+  omits the content-type header when `datacontenttype` is unset and stringifies
+  attribute values before encoding. ([#305])
+
 ## [2.2.0]
 
 ### Changed
@@ -368,3 +375,4 @@ CloudEvents v2 is a rewrite with ongoing development ([#271])
 [#279]: https://github.com/cloudevents/sdk-python/pull/279
 [#284]: https://github.com/cloudevents/sdk-python/pull/284
 [#291]: https://github.com/cloudevents/sdk-python/pull/291
+[#305]: https://github.com/cloudevents/sdk-python/pull/305
diff --git a/src/cloudevents/v1/kafka/conversion.py b/src/cloudevents/v1/kafka/conversion.py
index 6497dbc7..024e539f 100644
--- a/src/cloudevents/v1/kafka/conversion.py
+++ b/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())
diff --git a/tests/test_v1_compat/test_kafka_conversions.py b/tests/test_v1_compat/test_kafka_conversions.py
index da9f14d0..47a81ce6 100644
--- a/tests/test_v1_compat/test_kafka_conversions.py
+++ b/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)