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

GH-2935: Call afterSingletonsInitialized() in SBFM

Browse Source 
StreamsBuilderFactoryBean: Call afterSingletonsInitialized() programmatically in StreamsBuilderFactoryManager

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2935
This commit is contained in:
Soby Chacko
2024-04-10 14:06:08 -04:00
parent b736b4a8fa
commit 24556fd79e
1 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-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.
@@ -23,6 +23,7 @@ import java.util.Set;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.context.SmartLifecycle;
@@ -102,6 +103,9 @@ public class StreamsBuilderFactoryManager implements SmartLifecycle {
final List<ConsumerProperties> consumerProperties = bindingServicePropertiesPerSbfb.get(streamsBuilderFactoryBean);
final boolean autoStartupDisabledOnAtLeastOneConsumerBinding = consumerProperties.stream().anyMatch(consumerProperties1 -> !consumerProperties1.isAutoStartup());
if (!autoStartupDisabledOnAtLeastOneConsumerBinding) {
if (streamsBuilderFactoryBean instanceof SmartInitializingSingleton sbfb) {
sbfb.afterSingletonsInstantiated();
}
streamsBuilderFactoryBean.start();
this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean);
}