Honoring auto startup in Kafka Streams binder (#1127)
* Honoring auto startup in Kafka Streams binder When using Kafka Streams bineder, the processors are started unconditionally, i.e. autoStartup is always true by default. If spring.kafka.streams.auto-startup is set, then honor that as the auto-startup flag in the binder. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1126 * Addressing PR review comments Auto startup flag is honored per individual consumer bindings as well. * Addressing PR review comments * Making KafkaStreamsRegistry#kafkaStreams set to use a concurrent Set.
This commit is contained in:
Soby Chacko
@@ -1870,6 +1870,102 @@ When there are multiple bindings present on a single function, invoking these op
|
||||
This is because all the bindings on a single function are backed by the same `StreamsBuilderFactoryBean`.
|
||||
Therefore, for the function above, either `function-in-0` or `function-out-0` will work.
|
||||
|
||||
=== Manually starting Kafka Streams processors
|
||||
|
||||
Spring Cloud Stream Kafka Streams binder offers an abstraction called `StreamsBuilderFactoryManager` on top of the `StreamsBuilderFactoryBean` from Spring for Apache Kafka.
|
||||
This manager API is used for controlling the multiple `StreamsBuilderFactoryBean` per processor in a binder based application.
|
||||
Therefore, when using the binder, if you manually want to control the auto starting of the various `StreamsBuilderFactoryBean` objects in the application, you need to use `StreamsBuilderFactoryManager`.
|
||||
You can use the property `spring.kafka.streams.auto-startup` and set this to `false` in order to turn off auto starting of the processors.
|
||||
Then, in the application, you can use something as below to start the processors using `StreamsBuilderFactoryManager`.
|
||||
|
||||
```
|
||||
@Bean
|
||||
public ApplicationRunner runner(StreamsBuilderFactoryManager sbfm) {
|
||||
return args -> {
|
||||
sbfm.start();
|
||||
};
|
||||
}
|
||||
```
|
||||
|
||||
This feature is handy, when you want your application to start in the main thread and let Kafka Streams processors start separately.
|
||||
For example, when you have a large state store that needs to be restored, if the processors are started normally as is the default case, this may block your application to start.
|
||||
If you are using some sort of liveness probe mechanism (for example on Kubernetes), it may think that the application is down and attempt a restart.
|
||||
In order to correct this, you can set `spring.kafka.streams.auto-startup` to `false` and follow the approach above.
|
||||
|
||||
Keep in mind that, when using the Spring Cloud Stream binder, you are not directly dealing with `StreamsBuilderFactoryBean` from Spring for Apache Kafka, rather `StreamsBuilderFactoryManager`, as the `StreamsBuilderFactoryBean` objects are internally managed by the binder.
|
||||
|
||||
=== Manually starting Kafka Streams processors selectively
|
||||
|
||||
While the approach laid out above will unconditionally apply auto start `false` to all the Kafka Streams processors in the application through `StreamsBuilderFactoryManager`, it is often desirable that only individually selected Kafka Streams processors are not auto started.
|
||||
For instance, let us assume that you have three different functions (processors) in your application and for one of the processors, you do not want to start it as part of the application startup.
|
||||
Here is an example of such a situation.
|
||||
|
||||
```
|
||||
|
||||
@Bean
|
||||
public Function<KStream<?, ?>, KStream<?, ?>> process1() {
|
||||
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Consumer<KStream<?, ?>> process2() {
|
||||
|
||||
}
|
||||
|
||||
@Bean
|
||||
public BiFunction<KStream<?, ?>, KTable<?, ?>, KStream<?, ?>> process3() {
|
||||
|
||||
}
|
||||
|
||||
```
|
||||
|
||||
In this scenario above, if you set `spring.kafka.streams.auto-startup` to `false`, then none of the processors will auto start during the application startup.
|
||||
In that case, you have to programmatically start them as described above by calling `start()` on the underlying `StreamsBuilderFactoryManager`.
|
||||
However, if we have a use case to selectively disable only one processor, then you have to set `auto-startup` on the individual binding for that processor.
|
||||
Let us assume that we don't want our `process3` function to auto start.
|
||||
This is a `BiFunction` with two input bindings - `process3-in-0` and `process3-in-1`.
|
||||
In order to avoid auto start for this processor, you can pick any of these input bindings and set `auto-startup` on them.
|
||||
It does not matter which binding you pick; if you wish, you can set `auto-startup` to `false` on both of them, but one will be sufficient.
|
||||
Because they share the same factory bean, you don't have to set autoStartup to false on both bindings, but it probably makes sense to do so, for clarity.
|
||||
|
||||
Here is the Spring Cloud Stream property that you can use to disable auto startup for this processor.
|
||||
|
||||
```
|
||||
spring.cloud.stream.bindings.process3-in-0.consumer.auto-startup: false
|
||||
```
|
||||
|
||||
or
|
||||
|
||||
```
|
||||
spring.cloud.stream.bindings.process3-in-1.consumer.auto-startup: false
|
||||
```
|
||||
|
||||
Then, you can manually start the processor either using the REST endpoint or using the `BindingsEndpoint` API as shown below.
|
||||
For this, you need to ensure that you have the Spring Boot actuator dependency on the classpath.
|
||||
|
||||
```
|
||||
curl -d '{"state":"STARTED"}' -H "Content-Type: application/json" -X POST http://localhost:8080/actuator/bindings/process3-in-0
|
||||
```
|
||||
|
||||
or
|
||||
|
||||
```
|
||||
@Autowired
|
||||
BindingsEndpoint endpoint;
|
||||
|
||||
@Bean
|
||||
public ApplicationRunner runner() {
|
||||
return args -> {
|
||||
endpoint.changeState("process3-in-0", State.STARTED);
|
||||
};
|
||||
}
|
||||
```
|
||||
|
||||
See https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#binding_visualization_control[this section] from the reference docs for more details on this mechanism.
|
||||
|
||||
NOTE: When controlling the bindings by disabling `auto-startup` as described in this section, please note that this is only available for consumer bindings.
|
||||
In other words, if you use the producer binding, `process3-out-0`, that does not have any effect in terms of disabling the auto starting of the processor, although this producer binding uses the same `StreamsBuilderFactoryBean` as the consumer bindings.
|
||||
|
||||
=== Tracing using Spring Cloud Sleuth
|
||||
|
||||
When Spring Cloud Sleuth is on the classpath of a Spring Cloud Stream Kafka Streams binder based application, both its consumer and producer are automatically instrumented with tracing information.
|
||||
|
||||
@@ -161,6 +161,8 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
|
||||
//wrap the proxy created during the initial target type binding with real object (KTable)
|
||||
kTableWrapper.wrap((KTable<Object, Object>) table);
|
||||
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactoryPerBinding(input, streamsBuilderFactoryBean);
|
||||
this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
|
||||
bindingServiceProperties.getConsumerProperties(input));
|
||||
arguments[index] = table;
|
||||
}
|
||||
else if (parameterType.isAssignableFrom(GlobalKTable.class)) {
|
||||
@@ -173,6 +175,8 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
|
||||
//wrap the proxy created during the initial target type binding with real object (KTable)
|
||||
globalKTableWrapper.wrap((GlobalKTable<Object, Object>) table);
|
||||
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactoryPerBinding(input, streamsBuilderFactoryBean);
|
||||
this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
|
||||
bindingServiceProperties.getConsumerProperties(input));
|
||||
arguments[index] = table;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -398,8 +398,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
|
||||
KafkaStreamsBindingInformationCatalogue catalogue,
|
||||
KafkaStreamsRegistry kafkaStreamsRegistry,
|
||||
@Nullable KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics,
|
||||
@Nullable KafkaStreamsMicrometerListener listener) {
|
||||
return new StreamsBuilderFactoryManager(catalogue, kafkaStreamsRegistry, kafkaStreamsBinderMetrics, listener);
|
||||
@Nullable KafkaStreamsMicrometerListener listener, KafkaProperties kafkaProperties) {
|
||||
return new StreamsBuilderFactoryManager(catalogue, kafkaStreamsRegistry, kafkaStreamsBinderMetrics, listener, kafkaProperties);
|
||||
}
|
||||
|
||||
@Bean
|
||||
|
||||
@@ -54,6 +54,8 @@ public class KafkaStreamsBindingInformationCatalogue {
|
||||
|
||||
private final Map<String, StreamsBuilderFactoryBean> streamsBuilderFactoryBeanPerBinding = new HashMap<>();
|
||||
|
||||
private final Map<StreamsBuilderFactoryBean, List<ConsumerProperties>> consumerPropertiesPerSbfb = new HashMap<>();
|
||||
|
||||
private final Map<Object, ResolvableType> outboundKStreamResolvables = new HashMap<>();
|
||||
|
||||
private final Map<KStream<?, ?>, Serde<?>> keySerdeInfo = new HashMap<>();
|
||||
@@ -137,11 +139,19 @@ public class KafkaStreamsBindingInformationCatalogue {
|
||||
this.streamsBuilderFactoryBeanPerBinding.put(binding, streamsBuilderFactoryBean);
|
||||
}
|
||||
|
||||
void addConsumerPropertiesPerSbfb(StreamsBuilderFactoryBean streamsBuilderFactoryBean, ConsumerProperties consumerProperties) {
|
||||
this.consumerPropertiesPerSbfb.computeIfAbsent(streamsBuilderFactoryBean, k -> new ArrayList<>());
|
||||
this.consumerPropertiesPerSbfb.get(streamsBuilderFactoryBean).add(consumerProperties);
|
||||
}
|
||||
|
||||
public Map<StreamsBuilderFactoryBean, List<ConsumerProperties>> getConsumerPropertiesPerSbfb() {
|
||||
return this.consumerPropertiesPerSbfb;
|
||||
}
|
||||
|
||||
Map<String, StreamsBuilderFactoryBean> getStreamsBuilderFactoryBeanPerBinding() {
|
||||
return this.streamsBuilderFactoryBeanPerBinding;
|
||||
}
|
||||
|
||||
|
||||
void addOutboundKStreamResolvable(Object key, ResolvableType outboundResolvable) {
|
||||
this.outboundKStreamResolvables.put(key, outboundResolvable);
|
||||
}
|
||||
|
||||
@@ -529,6 +529,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
|
||||
this.kafkaStreamsBindingInformationCatalogue.addKeySerde((KStream<?, ?>) kStreamWrapper, keySerde);
|
||||
|
||||
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactoryPerBinding(input, streamsBuilderFactoryBean);
|
||||
this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
|
||||
bindingServiceProperties.getConsumerProperties(input));
|
||||
|
||||
if (KStream.class.isAssignableFrom(stringResolvableTypeMap.get(input).getRawClass())) {
|
||||
final Class<?> valueClass =
|
||||
|
||||
@@ -17,12 +17,12 @@
|
||||
package org.springframework.cloud.stream.binder.kafka.streams;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.apache.kafka.streams.KafkaStreams;
|
||||
import org.apache.kafka.streams.StreamsConfig;
|
||||
@@ -37,9 +37,9 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean;
|
||||
*/
|
||||
public class KafkaStreamsRegistry {
|
||||
|
||||
private Map<KafkaStreams, StreamsBuilderFactoryBean> streamsBuilderFactoryBeanMap = new HashMap<>();
|
||||
private final Map<KafkaStreams, StreamsBuilderFactoryBean> streamsBuilderFactoryBeanMap = new ConcurrentHashMap<>();
|
||||
|
||||
private final Set<KafkaStreams> kafkaStreams = new HashSet<>();
|
||||
private final Set<KafkaStreams> kafkaStreams = ConcurrentHashMap.newKeySet();
|
||||
|
||||
Set<KafkaStreams> getKafkaStreams() {
|
||||
Set<KafkaStreams> currentlyRunningKafkaStreams = new HashSet<>();
|
||||
|
||||
@@ -320,6 +320,9 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr
|
||||
this.kafkaStreamsBindingInformationCatalogue.registerBindingProperties(stream, bindingProperties1);
|
||||
|
||||
this.kafkaStreamsBindingInformationCatalogue.addStreamBuilderFactoryPerBinding(inboundName, streamsBuilderFactoryBean);
|
||||
this.kafkaStreamsBindingInformationCatalogue.addConsumerPropertiesPerSbfb(streamsBuilderFactoryBean,
|
||||
bindingServiceProperties.getConsumerProperties(inboundName));
|
||||
|
||||
for (StreamListenerParameterAdapter streamListenerParameterAdapter : adapters) {
|
||||
if (streamListenerParameterAdapter.supports(stream.getClass(),
|
||||
methodParameter)) {
|
||||
|
||||
@@ -16,11 +16,15 @@
|
||||
|
||||
package org.springframework.cloud.stream.binder.kafka.streams;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
|
||||
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
|
||||
import org.springframework.cloud.stream.binder.ConsumerProperties;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
import org.springframework.kafka.KafkaException;
|
||||
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
|
||||
@@ -40,7 +44,7 @@ import org.springframework.kafka.streams.KafkaStreamsMicrometerListener;
|
||||
*
|
||||
* @author Soby Chacko
|
||||
*/
|
||||
class StreamsBuilderFactoryManager implements SmartLifecycle {
|
||||
public class StreamsBuilderFactoryManager implements SmartLifecycle {
|
||||
|
||||
private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue;
|
||||
|
||||
@@ -52,19 +56,23 @@ class StreamsBuilderFactoryManager implements SmartLifecycle {
|
||||
|
||||
private volatile boolean running;
|
||||
|
||||
private final KafkaProperties kafkaProperties;
|
||||
|
||||
StreamsBuilderFactoryManager(KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue,
|
||||
KafkaStreamsRegistry kafkaStreamsRegistry,
|
||||
KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics,
|
||||
KafkaStreamsMicrometerListener listener) {
|
||||
KafkaStreamsRegistry kafkaStreamsRegistry,
|
||||
KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics,
|
||||
KafkaStreamsMicrometerListener listener,
|
||||
KafkaProperties kafkaProperties) {
|
||||
this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue;
|
||||
this.kafkaStreamsRegistry = kafkaStreamsRegistry;
|
||||
this.kafkaStreamsBinderMetrics = kafkaStreamsBinderMetrics;
|
||||
this.listener = listener;
|
||||
this.kafkaProperties = kafkaProperties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return true;
|
||||
return this.kafkaProperties == null || this.kafkaProperties.getStreams().isAutoStartup();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -81,19 +89,24 @@ class StreamsBuilderFactoryManager implements SmartLifecycle {
|
||||
try {
|
||||
Set<StreamsBuilderFactoryBean> streamsBuilderFactoryBeans = this.kafkaStreamsBindingInformationCatalogue
|
||||
.getStreamsBuilderFactoryBeans();
|
||||
int n = 0;
|
||||
for (StreamsBuilderFactoryBean streamsBuilderFactoryBean : streamsBuilderFactoryBeans) {
|
||||
if (this.listener != null) {
|
||||
streamsBuilderFactoryBean.addListener(this.listener);
|
||||
}
|
||||
// By default, we shutdown the client if there is an uncaught exception in the application.
|
||||
// By default, we shut down the client if there is an uncaught exception in the application.
|
||||
// Users can override this by customizing SBFB. See this issue for more details:
|
||||
// https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1110
|
||||
streamsBuilderFactoryBean.setStreamsUncaughtExceptionHandler(exception ->
|
||||
StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
|
||||
// Starting the stream.
|
||||
streamsBuilderFactoryBean.start();
|
||||
this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean);
|
||||
final Map<StreamsBuilderFactoryBean, List<ConsumerProperties>> bindingServicePropertiesPerSbfb =
|
||||
this.kafkaStreamsBindingInformationCatalogue.getConsumerPropertiesPerSbfb();
|
||||
final List<ConsumerProperties> consumerProperties = bindingServicePropertiesPerSbfb.get(streamsBuilderFactoryBean);
|
||||
final boolean autoStartupDisabledOnAtLeastOneConsumerBinding = consumerProperties.stream().anyMatch(consumerProperties1 -> !consumerProperties1.isAutoStartup());
|
||||
if (!autoStartupDisabledOnAtLeastOneConsumerBinding) {
|
||||
streamsBuilderFactoryBean.start();
|
||||
this.kafkaStreamsRegistry.registerKafkaStreams(streamsBuilderFactoryBean);
|
||||
}
|
||||
}
|
||||
if (this.kafkaStreamsBinderMetrics != null) {
|
||||
this.kafkaStreamsBinderMetrics.addMetrics(streamsBuilderFactoryBeans);
|
||||
|
||||
@@ -51,6 +51,7 @@ import org.springframework.cloud.stream.binder.Binding;
|
||||
import org.springframework.cloud.stream.binder.DefaultBinding;
|
||||
import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
|
||||
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsRegistry;
|
||||
import org.springframework.cloud.stream.binder.kafka.streams.StreamsBuilderFactoryManager;
|
||||
import org.springframework.cloud.stream.binder.kafka.streams.endpoint.KafkaStreamsTopologyEndpoint;
|
||||
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
|
||||
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
|
||||
@@ -232,6 +233,51 @@ public class KafkaStreamsBinderWordCountFunctionTests {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testKstreamBinderAutoStartup() throws Exception {
|
||||
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
|
||||
app.setWebApplicationType(WebApplicationType.NONE);
|
||||
|
||||
try (ConfigurableApplicationContext context = app.run(
|
||||
"--server.port=0",
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.kafka.streams.auto-startup=false",
|
||||
"--spring.cloud.stream.bindings.process-in-0.destination=words-3",
|
||||
"--spring.cloud.stream.bindings.process-out-0.destination=counts-3",
|
||||
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
|
||||
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
|
||||
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
|
||||
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
|
||||
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
|
||||
final StreamsBuilderFactoryManager streamsBuilderFactoryManager = context.getBean(StreamsBuilderFactoryManager.class);
|
||||
assertThat(streamsBuilderFactoryManager.isAutoStartup()).isFalse();
|
||||
assertThat(streamsBuilderFactoryManager.isRunning()).isFalse();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testKstreamIndividualBindingAutoStartup() throws Exception {
|
||||
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
|
||||
app.setWebApplicationType(WebApplicationType.NONE);
|
||||
|
||||
try (ConfigurableApplicationContext context = app.run(
|
||||
"--server.port=0",
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.bindings.process-in-0.destination=words-4",
|
||||
"--spring.cloud.stream.bindings.process-in-0.consumer.auto-startup=false",
|
||||
"--spring.cloud.stream.bindings.process-out-0.destination=counts-4",
|
||||
"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
|
||||
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
|
||||
"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
|
||||
"=org.apache.kafka.common.serialization.Serdes$StringSerde",
|
||||
"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
|
||||
final StreamsBuilderFactoryBean streamsBuilderFactoryBean = context.getBean(StreamsBuilderFactoryBean.class);
|
||||
assertThat(streamsBuilderFactoryBean.isRunning()).isFalse();
|
||||
streamsBuilderFactoryBean.start();
|
||||
assertThat(streamsBuilderFactoryBean.isRunning()).isTrue();
|
||||
}
|
||||
}
|
||||
|
||||
private void receiveAndValidate(String in, String out) {
|
||||
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
|
||||
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
|
||||
|
||||
Reference in New Issue
Block a user