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

GH-2857: Kafka Streams binder customization issues

Browse Source 
- StreamsBuilderFactoryBeanConfigurer customizaton happpens only after
   the StreamsBuilder object is created by the factory bean. This creates
   a scenario in which the customizations provided by the application are
   skipped by the StreamsBuilder. Addressing this issue.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2857
This commit is contained in:
Soby Chacko
2023-11-22 18:41:32 -05:00
parent 652b22cf76
commit d30f94af46
2 changed files with 9 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -344,14 +344,11 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
extendedConsumerProperties.setApplicationId((String) streamConfiguration.get(StreamsConfig.APPLICATION_ID_CONFIG));
final StreamsBuilderFactoryBean streamsBuilderFactoryBeanFromContext = applicationContext.getBean(
"&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class);
//At this point, the StreamsBuilderFactoryBean is created. If the users call, getObject()
//in the customizer, that should grant access to the StreamsBuilder.
if (customizer != null) {
customizer.configure(streamsBuilderFactoryBean);
}
return streamsBuilderFactoryBeanFromContext;
return applicationContext.getBean(
"&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class);
}
private void handleConcurrency(ApplicationContext applicationContext, String inboundName,

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java
View File

@@ -33,6 +33,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
@@ -58,6 +59,7 @@ import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.KafkaStreamsInfrastructureCustomizer;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@@ -409,11 +411,13 @@ class KafkaStreamsBinderWordCountFunctionTests {
return fb -> {
try {
fb.setStateListener((newState, oldState) -> {
});
fb.getObject(); //make sure no exception is thrown at this call.
fb.setInfrastructureCustomizer(new KafkaStreamsInfrastructureCustomizer() {
@Override
public void configureBuilder(StreamsBuilder builder) {
}
});
KafkaStreamsBinderWordCountFunctionTests.LATCH.countDown();
}
catch (Exception e) {
//Nothing to do - When the exception is thrown above, the latch won't be count down.