AbstractKafkaStreamsBinderProcessor ignores TimestampExtractor configuration starting with kafka-streams 3.7.0

[qxv1612]

Hello,

about this code snippet starting at AbstractKafkaStreamsBinderProcessor:627 (version 4.1.0):

	if (timestampExtractor != null) {
		consumed.withTimestampExtractor(timestampExtractor);
	}

"consumed" is an Consumed-instance which is part of kafka-streams.

In kafka-streams 3.6.1 a call to consumed.withTimestampExtractor() changes the existing instance.
Since kafka-streams 3.7.0 the same call creates a new instance of "Consumed", but the code above ignores this new instance.
In the end, the timestampExtractor is not used at all since kafka 3.7.0.

Same for "consumed.withName" a couple of lines below.

IMHO this is a big change in kafka-streams inside a minor version change.

To fix this anyway the code above might be changed to

	if (timestampExtractor != null) {
		consumed = consumed.withTimestampExtractor(timestampExtractor);
	}

Regards,

Ralf

[sobychacko]

@qxv1612 Your fix makes sense. If you are willing, please feel free to send a PR with the fix. Otherwise, we will look into it later. Thanks!

[qxv1612]

Hello @sobychacko,
I forked the repo and created the PR based on the main branch. I'm not sure if this is the right method.

A local build currently fails in KafkaBinderTests, but I expect its related to my local env settings.
Actually I started the same test with the current main branch.