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

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 https://github.com/spring-cloud/spring-cloud-stream/commit/d30044b25e13302472fb418de239d7ee1ffa756c .

	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.