V1 fix kafka datacontenttype issue in to_structured conversion

CHANGELOG.md

@@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.13.0]
+
+### Fixed
+- Kafka `to_structured` converter dropping `datacontentype` attribute
+  when setting a header. ([#296])
+
 ## [1.12.1]
 
 ### Changed
@@ -310,3 +316,4 @@ CloudEvents v2 is a rewrite with ongoing development ([#271])
 [#248]: https://github.com/cloudevents/sdk-python/pull/248
 [#249]: https://github.com/cloudevents/sdk-python/pull/249
 [#271]: https://github.com/cloudevents/sdk-python/pull/271
+[#296]: https://github.com/cloudevents/sdk-python/pull/296

cloudevents/__init__.py

@@ -12,4 +12,4 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-__version__ = "1.12.1"
+__version__ = "1.13.0"

cloudevents/kafka/conversion.py

@@ -213,8 +213,9 @@ def to_structured(
         attrs["data"] = data
 
     headers = {}
-    if "datacontenttype" in attrs:
-        headers["content-type"] = attrs.pop("datacontenttype").encode("utf-8")
+    datacontenttype = attrs.get("datacontenttype")
+    if datacontenttype is not None:
+        headers["content-type"] = datacontenttype.encode("utf-8")
 
     try:
         value = envelope_marshaller(attrs)

cloudevents/tests/test_kafka_conversions.py

@@ -247,6 +247,7 @@ def test_sets_value_default_marshallers(self, source_event):
                 "source": source_event["source"],
                 "type": source_event["type"],
                 "time": source_event["time"],
+                "datacontenttype": source_event["datacontenttype"],
                 "partitionkey": source_event["partitionkey"],
                 "data": self.expected_data,
             }
@@ -263,6 +264,7 @@ def test_sets_value_custom_data_marshaller_default_envelope(
                 "source": source_event["source"],
                 "type": source_event["type"],
                 "time": source_event["time"],
+                "datacontenttype": source_event["datacontenttype"],
                 "partitionkey": source_event["partitionkey"],
                 "data_base64": base64.b64encode(
                     custom_marshaller(self.expected_data)
@@ -281,6 +283,7 @@ def test_sets_value_custom_envelope_marshaller(
                 "source": source_event["source"],
                 "type": source_event["type"],
                 "time": source_event["time"],
+                "datacontenttype": source_event["datacontenttype"],
                 "partitionkey": source_event["partitionkey"],
                 "data": self.expected_data,
             }
@@ -299,6 +302,7 @@ def test_sets_value_custom_marshallers(self, source_event, custom_marshaller):
                 "source": source_event["source"],
                 "type": source_event["type"],
                 "time": source_event["time"],
+                "datacontenttype": source_event["datacontenttype"],
                 "partitionkey": source_event["partitionkey"],
                 "data_base64": base64.b64encode(
                     custom_marshaller(self.expected_data)
@@ -335,6 +339,13 @@ def test_sets_headers(self, source_event):
             "utf-8"
         )
 
+    def test_datacontenttype_attribute_present_after_setting_header(self, source_event):
+        result = to_structured(source_event)
+        datacontenttype = source_event.get("datacontenttype")
+        assert len(result.headers) == 1
+        assert result.headers["content-type"] == datacontenttype.encode("utf-8")
+        assert datacontenttype in result.value.decode("utf-8")
+
     def test_datamarshaller_exception(self, source_event):
         with pytest.raises(cloud_exceptions.DataMarshallerError):
             to_structured(source_event, data_marshaller=failing_func)