v4.3.3 no longer adds a header "TIMESTAMP" on kafka messages consumed

[aDecoy]

version 4.3.0 used to add a "timestamp" header when consumming kafka messages. version 4.3.3 does not. We used this value and it seemed like a breaking change when it disapeared on a patch bump.

The reason is in how org.springframework.cloud.stream.function.FunctionConfiguration#sanitize was changed in the commit d30044b .

	private static <P> Message<P> sanitize(Message<P> inputMessage) {
		return MessageBuilder
			.fromMessage(inputMessage)
			.removeHeader("spring.cloud.stream.sendto.destination")
			.setHeader(MessageUtils.SOURCE_TYPE, inputMessage.getHeaders().get(MessageUtils.TARGET_PROTOCOL))
			.removeHeader(MessageUtils.TARGET_PROTOCOL)
			.build();
	}

was changed to :

	private static <P> Message<P> sanitize(Message<P> inputMessage) {
		return MessageBuilder
			.fromMessage(inputMessage)
			.removeHeader("spring.cloud.stream.sendto.destination")
		//	.setHeader(MessageUtils.SOURCE_TYPE, inputMessage.getHeaders().get(MessageUtils.TARGET_PROTOCOL))
		//	.removeHeader(MessageUtils.TARGET_PROTOCOL)
			.build();
	}

The change caused an .build() to behave different since headerAccessor.isModified() now returns false. The builder has an if that has this as a predicate.

public Message<T> build() {
		if (!this.modified && !this.headerAccessor.isModified() && this.originalMessage != null
				&& !containsReadOnly(this.originalMessage.getHeaders())) {

			return this.originalMessage;
		}
		if (this.payload instanceof Throwable throwable) {
			return (Message<T>) new ErrorMessage(throwable, this.headerAccessor.toMap());
		}
		return new GenericMessage<>(this.payload, this.headerAccessor.toMap());
	}

org.springframework.integration.support.BaseMessageBuilder#build would thus return originalMessage instead of using the GenericMessage constructor. And only GenericMessage constructor added a "timestamp" header.

location outside of your repo: (org/springframework/spring-messaging/6.2.18/spring-messaging-6.2.18-sources.jar!/org/springframework/messaging/MessageHeaders.java:151)

To Reproduce
Steps to reproduce the behavior:

1. Consume a kafka message with a consumer

public class ValidationConsumer implements Consumer<Message<DataBlock>> {

    @Override
    public void accept(Message<DataBlock> message) {
        var messageHeaders = message.getHeaders(); 
 System.out.print(messageHeaders.getTimestamp() )  // v4.3.0 has a value here, v4.3.3 has null

Additional context
We found this when bumping spring-cloud-dependencies from 2025.0.0 to 2025.0.3.

We dont need a quick fix for this. We will change our application code to use our own start time instead of relying on the "timstamp" header that used to be provided by this library. It feels more robust to use our own startime for our own application needs.

[PRAHLAD09-dev]

I looked into this — confirmed the root cause is in FunctionConfiguration#sanitize(). Since the two setHeader/removeHeader calls were commented out, headerAccessor.isModified() returns false, so MessageBuilder.build() takes the fast-path and returns originalMessage directly instead of constructing a new GenericMessage — and only the GenericMessage constructor stamps a timestamp. I'd like to look at the original commit's intent (d30044b) before proposing a fix, so I don't regress whatever it was solving. Happy to send a PR — let me know if that direction works.