Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-621: Add KafkaStreams customizer

Browse Source 
Fixes spring-projects/spring-kafka#621

Refactor;
- Revert configurer
- Add setters to `StreamsBuilderFactoryBean`
- Create `CompositeKafkaStreamsCustomizer`

Polish, update docs.

* Polishing code style, Docs and Java Docs
This commit is contained in:
Nurettin Yilmaz
2018-03-27 02:36:29 +03:00
committed by Artem Bilan
parent 202f73cfa4
commit e77e1efbc5
5 changed files with 243 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
spring-kafka/src/main/java/org/springframework/kafka/core/CompositeKafkaStreamsCustomizer.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* Copyright 2018 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.kafka.core;
import java.util.ArrayList;
import java.util.List;
import org.apache.kafka.streams.KafkaStreams;
/**
* Composite {@link KafkaStreamsCustomizer} customizes {@link KafkaStreams} by delegating
* to a list of provided {@link KafkaStreamsCustomizer}.
*
* @author Nurettin Yilmaz
* @author Artem Bilan
*
* @since 2.1.5
*/
public class CompositeKafkaStreamsCustomizer implements KafkaStreamsCustomizer {
private final List<KafkaStreamsCustomizer> kafkaStreamsCustomizers = new ArrayList<>();
public CompositeKafkaStreamsCustomizer() {
}
public CompositeKafkaStreamsCustomizer(List<KafkaStreamsCustomizer> kafkaStreamsCustomizers) {
this.kafkaStreamsCustomizers.addAll(kafkaStreamsCustomizers);
}
@Override
public void customize(KafkaStreams kafkaStreams) {
this.kafkaStreamsCustomizers.forEach(kafkaStreamsCustomizer -> kafkaStreamsCustomizer.customize(kafkaStreams));
}
public void addKafkaStreamsCustomizers(List<KafkaStreamsCustomizer> kafkaStreamsCustomizers) {
this.kafkaStreamsCustomizers.addAll(kafkaStreamsCustomizers);
}
}

35
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaStreamsCustomizer.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* Copyright 2018 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.kafka.core;
import org.apache.kafka.streams.KafkaStreams;
/**
* Callback interface that can be used to configure {@link KafkaStreams} directly.
*
* @author Nurettin Yilmaz
*
* @since 2.1.5
*
* @see StreamsBuilderFactoryBean
*/
@FunctionalInterface
public interface KafkaStreamsCustomizer {
void customize(KafkaStreams kafkaStreams);
}

54
spring-kafka/src/main/java/org/springframework/kafka/core/StreamsBuilderFactoryBean.java
View File

@@ -23,6 +23,8 @@ import org.apache.kafka.streams.KafkaClientSupplier;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
import org.springframework.beans.factory.config.AbstractFactoryBean;
@@ -35,10 +37,14 @@ import org.springframework.util.Assert;
* An {@link AbstractFactoryBean} for the {@link StreamsBuilder} instance
* and lifecycle control for the internal {@link KafkaStreams} instance.
*
* <p>A fine grained control on {@link KafkaStreams} can be achieved by
* {@link KafkaStreamsCustomizer}s</p>
*
* @author Artem Bilan
* @author Ivan Ursul
* @author Soby Chacko
* @author Zach Olauson
* @author Nurettin Yilmaz
*
* @since 1.1.4
*/
@@ -46,24 +52,28 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
private static final int DEFAULT_CLOSE_TIMEOUT = 10;
private final CleanupConfig cleanupConfig;
private KafkaClientSupplier clientSupplier = new DefaultKafkaClientSupplier();
private StreamsConfig streamsConfig;
private KafkaStreams kafkaStreams;
private final CleanupConfig cleanupConfig;
private KafkaClientSupplier clientSupplier = new DefaultKafkaClientSupplier();
private KafkaStreamsCustomizer kafkaStreamsCustomizer;
private KafkaStreams.StateListener stateListener;
private StateRestoreListener stateRestoreListener;
private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
private boolean autoStartup = true;
private int phase = Integer.MAX_VALUE - 1000;
private KafkaStreams.StateListener stateListener;
private Thread.UncaughtExceptionHandler exceptionHandler;
private int closeTimeout = DEFAULT_CLOSE_TIMEOUT;
private KafkaStreams kafkaStreams;
private volatile boolean running;
/**
@@ -140,12 +150,27 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
this.clientSupplier = clientSupplier; // NOSONAR (sync)
}
/**
* Specify a {@link KafkaStreamsCustomizer} to customize a {@link KafkaStreams}
* instance during {@link #start()}.
* @param kafkaStreamsCustomizer the {@link KafkaStreamsCustomizer} to use.
* @since 2.1.5
*/
public void setKafkaStreamsCustomizer(KafkaStreamsCustomizer kafkaStreamsCustomizer) {
Assert.notNull(kafkaStreamsCustomizer, "'kafkaStreamsCustomizer' must not be null");
this.kafkaStreamsCustomizer = kafkaStreamsCustomizer; // NOSONAR (sync)
}
public void setStateListener(KafkaStreams.StateListener stateListener) {
this.stateListener = stateListener; // NOSONAR (sync)
}
public void setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler exceptionHandler) {
this.exceptionHandler = exceptionHandler; // NOSONAR (sync)
this.uncaughtExceptionHandler = exceptionHandler; // NOSONAR (sync)
}
public void setStateRestoreListener(StateRestoreListener stateRestoreListener) {
this.stateRestoreListener = stateRestoreListener; // NOSONAR (sync)
}
/**
@@ -171,7 +196,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
return new StreamsBuilder();
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
@@ -198,9 +222,17 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
if (!this.running) {
try {
Assert.notNull(this.streamsConfig, "'streamsConfig' must not be null");
this.kafkaStreams = new KafkaStreams(getObject().build(), this.streamsConfig, this.clientSupplier);
Topology topology = getObject().build();
if (logger.isDebugEnabled()) {
logger.debug(topology.describe());
}
this.kafkaStreams = new KafkaStreams(topology, this.streamsConfig, this.clientSupplier);
this.kafkaStreams.setStateListener(this.stateListener);
this.kafkaStreams.setUncaughtExceptionHandler(this.exceptionHandler);
this.kafkaStreams.setGlobalStateRestoreListener(this.stateRestoreListener);
this.kafkaStreams.setUncaughtExceptionHandler(this.uncaughtExceptionHandler);
if (this.kafkaStreamsCustomizer != null) {
this.kafkaStreamsCustomizer.customize(this.kafkaStreams);
}
if (this.cleanupConfig.cleanupOnStart()) {
this.kafkaStreams.cleanUp();
}

109
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaStreamsCustomizerTests.java Normal file
View File

@@ -0,0 +1,109 @@
/*
* Copyright 2018 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.kafka.core;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Nurettin Yilmaz
*
* @since 2.1.5
*/
@SpringJUnitConfig
@DirtiesContext
@EmbeddedKafka
public class KafkaStreamsCustomizerTests {
private static final String APPLICATION_ID = "testStreams";
private static final TestStateListener STATE_LISTENER = new TestStateListener();
@Autowired
private StreamsBuilderFactoryBean streamsBuilderFactoryBean;
@Test
public void testKafkaStreamsCustomizer() {
KafkaStreams.State state = this.streamsBuilderFactoryBean.getKafkaStreams().state();
assertThat(STATE_LISTENER.getCurrentState()).isEqualTo(state);
}
@Configuration
@EnableKafka
@EnableKafkaStreams
public static class KafkaStreamsConfiguration {
@Value("${" + KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
private String brokerAddresses;
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder() {
StreamsBuilderFactoryBean streamsBuilderFactoryBean = new StreamsBuilderFactoryBean(kStreamsConfigs());
streamsBuilderFactoryBean.setKafkaStreamsCustomizer(customizer());
return streamsBuilderFactoryBean;
}
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public StreamsConfig kStreamsConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
return new StreamsConfig(props);
}
private KafkaStreamsCustomizer customizer() {
return kafkaStreams -> kafkaStreams.setStateListener(STATE_LISTENER);
}
}
static class TestStateListener implements KafkaStreams.StateListener {
private KafkaStreams.State currentState;
@Override
public void onChange(KafkaStreams.State newState, KafkaStreams.State oldState) {
this.currentState = newState;
}
KafkaStreams.State getCurrentState() {
return this.currentState;
}
}
}

4
src/reference/asciidoc/streams.adoc
View File

@@ -70,7 +70,9 @@ If you would like to control lifecycle manually (e.g. stop and start by some con
Since `StreamsBuilderFactoryBean` utilize its internal `KafkaStreams` instance, it is safe to stop and restart it again - a new `KafkaStreams` is created on each `start()`.
Also consider using different `StreamsBuilderFactoryBean` s, if you would like to control lifecycles for `KStream` instances separately.
You can specify `KafkaStreams.StateListener` and `Thread.UncaughtExceptionHandler` options on the `StreamsBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance.
You also can specify `KafkaStreams.StateListener`, `Thread.UncaughtExceptionHandler` and `StateRestoreListener` options on the `StreamsBuilderFactoryBean` which are delegated to the internal `KafkaStreams` instance.
Also apart from setting those options indirectly on `StreamsBuilderFactoryBean`, starting with _version 2.1.5_, a `KafkaStreamsCustomizer` callback interface can be used to configure inner `KafkaStreams` instance.
Note that `KafkaStreamsCustomizer` will override the options which are given via `StreamsBuilderFactoryBean`.
That internal `KafkaStreams` instance can be accessed via `StreamsBuilderFactoryBean.getKafkaStreams()` if you need to perform some `KafkaStreams` operations directly.
You can autowire `StreamsBuilderFactoryBean` bean by type, but you should be sure that you use full type in the bean definition, for example: