Description

When using CloudEventMessageBuilder.setDataContentType(...) to produce a CloudEvent in binary content mode to Kafka, the resulting Kafka record header is ce_datacontenttype instead of the spec-required content-type.

Root Cause

setDataContentType() stores the value under the key CloudEventMessageUtils.DATACONTENTTYPE, which resolves to "ce-datacontenttype" (DEFAULT_ATTR_PREFIX + "datacontenttype"). When Spring Cloud Stream's Kafka binder maps headers to the Kafka record, the ce- prefix is translated to ce_, resulting in the header ce_datacontenttype.

However, the CloudEvents Kafka Protocol Binding spec (section 3.2.1) states:

For the binary mode, the header content-type property MUST be mapped directly to the CloudEvents datacontenttype attribute.

The spec example confirms this — content-type is used as a plain header without the ce_ prefix:

------------------ headers -------------------
ce_specversion: "1.0"
ce_type: "com.example.someevent"
ce_source: "/mycontext/subcontext"
ce_id: "1234-1234-1234"
ce_time: "2018-04-05T03:56:24Z"
content-type: application/avro

Expected Behavior

When writing a CloudEvent in binary mode to Kafka, datacontenttype should be written as the content-type header (no ce_ prefix).

Actual Behavior

The value is written as ce_datacontenttype, which violates the CloudEvents Kafka binding spec. Consumers following the spec will not find the content type where they expect it.

Impact

• Interoperability with non-Spring CloudEvents consumers is broken
• Consumers that check content-type to distinguish structured vs binary mode (as per spec section 3) will not detect the content type correctly

Suggested Fix

The Kafka header mapping logic should special-case datacontenttype to write it as the plain content-type header rather than applying the standard ce_ prefix transformation. This is analogous to how the HTTP binding maps datacontenttype to the Content-Type HTTP header without the ce- prefix.