GH-1401 Use content-type for Kafka CloudEvents

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventMessageBuilder.java

@@ -186,7 +186,13 @@ else if (key.startsWith(CloudEventMessageUtils.KAFKA_ATTR_PREFIX)) {
 	private void swapPrefix(String key, String currentPrefix, String newPrefix) {
 		Object value = headers.remove(key);
 		key = key.substring(currentPrefix.length());
-		this.headers.put(newPrefix + key, value);
+		if (newPrefix.equals(CloudEventMessageUtils.KAFKA_ATTR_PREFIX)
+				&& key.equals(CloudEventMessageUtils._DATACONTENTTYPE)) {
+			this.headers.put("content-type", value);
+		}
+		else {
+			this.headers.put(newPrefix + key, value);
+		}
 	}
 
 	private Message<T> doBuild(String prefix) {

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventMessageUtils.java

@@ -205,10 +205,16 @@ public static String getType(Message<?> message) {
 	public static String getDataContentType(Message<?> message) {
 		String prefix = determinePrefixToUse(message.getHeaders());
 		Object value = message.getHeaders().get(prefix + _DATACONTENTTYPE);
+		if (value == null && isCloudEvent(message)) {
+			value = message.getHeaders().get("content-type");
+			if (value == null) {
+				value = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
+			}
+		}
 		if (value instanceof byte[] v) {
 			value = toString(v);
 		}
-		return (String) value;
+		return value != null ? value.toString() : null;
 	}
 
 	public static URI getDataSchema(Message<?> message) {

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/cloudevent/CloudEventMessageUtilsAndBuilderTests.java

@@ -98,6 +98,30 @@ public void testAttributeRecognitionAndCanonicalization() {
 		assertThat(CloudEventMessageUtils.getSpecVersion(httpMessage)).isEqualTo("1.0");
 	}
 
+	@Test
+	void buildWithKafkaPrefixUsesContentTypeHeaderForDataContentType() {
+		Message<String> kafkaMessage = CloudEventMessageBuilder.withData("hello")
+				.setDataContentType("application/json")
+				.build(CloudEventMessageUtils.KAFKA_ATTR_PREFIX);
+
+		assertThat(kafkaMessage.getHeaders())
+				.containsEntry("content-type", "application/json")
+				.doesNotContainKey("ce_datacontenttype");
+		assertThat(CloudEventMessageUtils.getDataContentType(kafkaMessage)).isEqualTo("application/json");
+	}
+
+	@Test
+	void buildWithDefaultPrefixUsesCloudEventDataContentTypeAttribute() {
+		Message<String> message = CloudEventMessageBuilder.withData("hello")
+				.setDataContentType("application/json")
+				.build(CloudEventMessageUtils.DEFAULT_ATTR_PREFIX);
+
+		assertThat(message.getHeaders())
+				.containsEntry("ce-datacontenttype", "application/json")
+				.doesNotContainKey("content-type");
+		assertThat(CloudEventMessageUtils.getDataContentType(message)).isEqualTo("application/json");
+	}
+
 	@Test
 	void canonicalizeHeadersWithPossibleCopyReturnsCopyWithUpdatedHeadersWhenModified() {
 		// TODO add the following test cases