Skip to content

ce_ prefix is no longer applied automatically for Cloud Events sent to Kafka #1461

Description

@DidierLoiseau

Describe the bug
Since 5.0, following #1294 (and spring-cloud/spring-cloud-stream#2222), it appears that Spring Cloud Function no longer determines automatically the correct ce_ Cloud Event prefix to use when a message is sent to Kafka. It only works properly when at least one of the headers has the kafka_ prefix (e.g. when using constants from org.springframework.kafka.support.KafkaHeaders), but if no such header is used, it falls back on the default ce- prefix.

Sample
This can be reproduced easily with function-sample-cloudevent-stream’s DemoApplicationTests.

According to the README (and also this Spring blog post which I think is the same):

Also note that on the receiving end Cloud Events attributes are now prefixed with ce_ prefix (see Cloud Events Kafka protocol bindings), since it was determined by the framework that the target destination is Apache Kafka.

However, when I run that test, I see it uses the cloudevents: prefix instead:

Result Message: GenericMessage [payload={"person":{"firstName":"John","lastName":"Doe"},"id":884,"message":"Employee 884 was hired on 04-09-2026"}, headers={cloudEvents:id=e790a03d-1033-4a9a-a69b-f40727209fc1, cloudEvents:type=io.spring.cloudevent.Employee, kafka_timestampType=CREATE_TIME, kafka_receivedTopic=hire-out-0, kafka_offset=2, cloudEvents:specversion=1.0, message-type=cloudevent, kafka_consumer=org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer@270ab7bc, cloudEvents:source=http://spring.io/, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTimestamp=1788535901863, kafka_groupId=test}]

and also from Kafkbat UI, I can see that the Kafka headers use the wrong prefix:

{
	"cloudEvents:id": "1c75b58c-b653-4713-a921-258d02ad331d",
	"spring_json_header_types": "{\"cloudEvents:id\":\"java.lang.String\",\"cloudEvents:type\":\"java.lang.String\",\"cloudEvents:specversion\":\"java.lang.String\",\"message-type\":\"java.lang.String\",\"cloudEvents:source\":\"java.lang.String\",\"contentType\":\"java.lang.String\"}",
	"cloudEvents:type": "io.spring.cloudevent.Employee",
	"cloudEvents:specversion": "1.0",
	"message-type": "cloudevent",
	"cloudEvents:source": "http://spring.io/",
	"contentType": "application/json"
}

As a workaround, when using CloudEventMessageBuilder it should be possible to explicitly add a kafka_ header, or to call build() with the right prefix, but then the code is no longer agnostic vs. the destination.

When I read spring-cloud/spring-cloud-stream#2222, it is unclear whether this workaround is the intended solution.

p.s. there are no release notes for 5.0.0?

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions