From 24556fd79e8d3afc66b30921cb42f0195ee3f10a Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 10 Apr 2024 14:06:08 -0400 Subject: [PATCH] GH-2935: Call afterSingletonsInitialized() in SBFM StreamsBuilderFactoryBean: Call afterSingletonsInitialized() programmatically in StreamsBuilderFactoryManager Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2935 --- .../binder/kafka/streams/StreamsBuilderFactoryManager.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index 3798345a2..fb59e7dd8 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -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 = 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); }