SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2922: Timestamp extractor - Kafka Streams 3.7.0

Browse Source 
* Address immutability changes for the call to `Consumed#withTimestampExtractor`.

 * In 3.7.0, this call returns a new instance of `Consumed` as oppposed to
   mutating the existing instance in the previous versions. Address this
   change in behavior in the Kafka Streams binder.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2922
This commit is contained in:
WOODMARK\r.wiedmann.extern
2024-03-20 08:05:10 +01:00
committed by Soby Chacko
parent 9190e9b835
commit eab86f81a8
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2023 the original author or authors.
* Copyright 2019-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -90,6 +90,7 @@ import org.springframework.util.StringUtils;
/**
* @author Soby Chacko
* @author Ralf Wiedmann
* @since 3.0.0
*/
public abstract class AbstractKafkaStreamsBinderProcessor implements ApplicationContextAware {
@@ -622,13 +623,13 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
timestampExtractor = applicationContext.getBean(kafkaStreamsConsumerProperties.getTimestampExtractorBeanName(),
TimestampExtractor.class);
}
final Consumed<K, V> consumed = Consumed.with(keySerde, valueSerde)
Consumed<K, V> consumed = Consumed.with(keySerde, valueSerde)
.withOffsetResetPolicy(autoOffsetReset);
if (timestampExtractor != null) {
consumed.withTimestampExtractor(timestampExtractor);
consumed = consumed.withTimestampExtractor(timestampExtractor);
}
if (StringUtils.hasText(kafkaStreamsConsumerProperties.getConsumedAs())) {
consumed.withName(kafkaStreamsConsumerProperties.getConsumedAs());
consumed = consumed.withName(kafkaStreamsConsumerProperties.getConsumedAs());
}
return consumed;
}