diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index de4a82990..1857c8dcc 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -942,7 +942,22 @@ public KStream process(KStream input) { === State Store State store is created automatically by Kafka Streams when the DSL is used. -When processor API is used, you need to register a state store manually. In order to do so, you can use `KafkaStreamsStateStore` annotation. +When processor API is used, you need to register a state store manually. In order to do so, you can create the StateStore as a bean in the application. +Here is an example of defining such a bean. + +``` +@Bean +public StoreBuilder mystore() { + return Stores.windowStoreBuilder( + Stores.persistentWindowStore("mystate", + 3L, 3, 3L, false), Serdes.String(), + Serdes.String()); +} +``` + +During the bootstrap, the above bean will be processed by the binder and pass on to the Streams builder object. +Defining custom state stores by providing them as beans is the preferred approach. +However, you can also use `KafkaStreamsStateStore` annotation for this. You can specify the name and type of the store, flags to control log and disabling cache, etc. Once the store is created by the binder during the bootstrapping phase, you can access this state store through the processor API. Below are some primitives for doing this. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index 9c7a7e4e1..4e11e0269 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; +import java.util.Arrays; import java.util.Map; import java.util.Properties; @@ -33,6 +34,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.state.KeyValueStore; +import org.apache.kafka.streams.state.StoreBuilder; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanDefinition; @@ -300,4 +302,45 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application return Serdes.ByteArray(); } } + + protected void addStateStoreBeans(StreamsBuilder streamsBuilder) { + try { + final Map storeBuilders = applicationContext.getBeansOfType(StoreBuilder.class); + if (!CollectionUtils.isEmpty(storeBuilders)) { + storeBuilders.values().forEach(storeBuilder -> { + streamsBuilder.addStateStore(storeBuilder); + if (LOG.isInfoEnabled()) { + LOG.info("state store " + storeBuilder.name() + " added to topology"); + } + }); + } + } + catch (Exception e) { + // Pass through. + } + } + + protected KStream getKStream(String inboundName, BindingProperties bindingProperties, StreamsBuilder streamsBuilder, + Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { + addStateStoreBeans(streamsBuilder); + + String[] bindingTargets = StringUtils.commaDelimitedListToStringArray( + this.bindingServiceProperties.getBindingDestination(inboundName)); + + KStream stream = streamsBuilder.stream(Arrays.asList(bindingTargets), + Consumed.with(keySerde, valueSerde) + .withOffsetResetPolicy(autoOffsetReset)); + final boolean nativeDecoding = this.bindingServiceProperties + .getConsumerProperties(inboundName).isUseNativeDecoding(); + if (nativeDecoding) { + LOG.info("Native decoding is enabled for " + inboundName + + ". Inbound deserialization done at the broker."); + } + else { + LOG.info("Native decoding is disabled for " + inboundName + + ". Inbound message conversion done by Spring Cloud Stream."); + } + + return getkStream(bindingProperties, stream, nativeDecoding); + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index 65de5b1ce..3834e05fc 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.ArrayList; -import java.util.Arrays; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; @@ -37,9 +36,7 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.Topology; -import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.state.StoreBuilder; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -60,7 +57,6 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.CleanupConfig; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; -import org.springframework.util.StringUtils; /** * @author Soby Chacko @@ -311,8 +307,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro final Topology.AutoOffsetReset autoOffsetReset = getAutoOffsetReset(input, extendedConsumerProperties); if (parameterType.isAssignableFrom(KStream.class)) { - KStream stream = getkStream(input, bindingProperties, - streamsBuilder, keySerde, valueSerde, autoOffsetReset); + KStream stream = getKStream(input, bindingProperties, streamsBuilder, keySerde, valueSerde, autoOffsetReset); KStreamBoundElementFactory.KStreamWrapper kStreamWrapper = (KStreamBoundElementFactory.KStreamWrapper) targetBean; //wrap the proxy created during the initial target type binding with real object (KStream) @@ -361,40 +356,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro BindingProperties bindingProperties, StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { - try { - final Map storeBuilders = applicationContext.getBeansOfType(StoreBuilder.class); - if (!CollectionUtils.isEmpty(storeBuilders)) { - storeBuilders.values().forEach(storeBuilder -> { - streamsBuilder.addStateStore(storeBuilder); - if (LOG.isInfoEnabled()) { - LOG.info("state store " + storeBuilder.name() + " added to topology"); - } - }); - } - } - catch (Exception e) { - // Pass through. - } - - String[] bindingTargets = StringUtils - .commaDelimitedListToStringArray(this.bindingServiceProperties.getBindingDestination(inboundName)); - - KStream stream = - streamsBuilder.stream(Arrays.asList(bindingTargets), - Consumed.with(keySerde, valueSerde) - .withOffsetResetPolicy(autoOffsetReset)); - final boolean nativeDecoding = this.bindingServiceProperties.getConsumerProperties(inboundName) - .isUseNativeDecoding(); - if (nativeDecoding) { - LOG.info("Native decoding is enabled for " + inboundName + ". " + - "Inbound deserialization done at the broker."); - } - else { - LOG.info("Native decoding is disabled for " + inboundName + ". " + - "Inbound message conversion done by Spring Cloud Stream."); - } - - return getkStream(bindingProperties, stream, nativeDecoding); + return getKStream(inboundName, bindingProperties, streamsBuilder, keySerde, valueSerde, autoOffsetReset); } @Override diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index e471fa361..094f6f1a1 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.lang.reflect.Method; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; import java.util.HashMap; import java.util.List; @@ -31,7 +30,6 @@ import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.Topology; -import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; @@ -371,24 +369,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr LOG.info("state store " + storeBuilder.name() + " added to topology"); } } - String[] bindingTargets = StringUtils.commaDelimitedListToStringArray( - this.bindingServiceProperties.getBindingDestination(inboundName)); - - KStream stream = streamsBuilder.stream(Arrays.asList(bindingTargets), - Consumed.with(keySerde, valueSerde) - .withOffsetResetPolicy(autoOffsetReset)); - final boolean nativeDecoding = this.bindingServiceProperties - .getConsumerProperties(inboundName).isUseNativeDecoding(); - if (nativeDecoding) { - LOG.info("Native decoding is enabled for " + inboundName - + ". Inbound deserialization done at the broker."); - } - else { - LOG.info("Native decoding is disabled for " + inboundName - + ". Inbound message conversion done by Spring Cloud Stream."); - } - - return getkStream(bindingProperties, stream, nativeDecoding); + return getKStream(inboundName, bindingProperties, streamsBuilder, keySerde, valueSerde, autoOffsetReset); } private void validateStreamListenerMethod(StreamListener streamListener, diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java index 431a24637..32741fdc0 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java @@ -18,9 +18,12 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration; import java.util.Map; +import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.state.StoreBuilder; +import org.apache.kafka.streams.state.Stores; import org.apache.kafka.streams.state.WindowStore; import org.junit.ClassRule; import org.junit.Test; @@ -34,12 +37,14 @@ import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsStateStore; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsStateStoreProperties; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; +import static junit.framework.TestCase.fail; import static org.assertj.core.api.Assertions.assertThat; /** @@ -73,7 +78,7 @@ public class KafkaStreamsStateStoreIntegrationTests { + embeddedKafka.getBrokersAsString()); try { Thread.sleep(2000); - receiveAndValidateFoo(context); + receiveAndValidateFoo(context, ProductCountApplication.class); } catch (Exception e) { throw e; @@ -83,13 +88,43 @@ public class KafkaStreamsStateStoreIntegrationTests { } } + @Test + public void testKstreamStateStoreBuilderBeansDefinedInApplication() throws Exception { + SpringApplication app = new SpringApplication(StateStoreBeanApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input3.destination=foobar", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--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.bindings.input3.consumer.applicationId" + + "=KafkaStreamsStateStoreIntegrationTests-xyzabc-123", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString()); + try { + Thread.sleep(2000); + receiveAndValidateFoo(context, StateStoreBeanApplication.class); + } + catch (Exception e) { + throw e; + } + finally { + context.close(); + } + } + + @Test public void testSameStateStoreIsCreatedOnlyOnceWhenMultipleInputBindingsArePresent() throws Exception { SpringApplication app = new SpringApplication(ProductCountApplicationWithMultipleInputBindings.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.destination=foobar", + "--spring.cloud.stream.bindings.input1.destination=foobar", + "--spring.cloud.stream.bindings.input2.destination=hello-foobar", "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + "=org.apache.kafka.common.serialization.Serdes$StringSerde", @@ -115,7 +150,7 @@ public class KafkaStreamsStateStoreIntegrationTests { } } - private void receiveAndValidateFoo(ConfigurableApplicationContext context) + private void receiveAndValidateFoo(ConfigurableApplicationContext context, Class clazz) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( @@ -126,13 +161,28 @@ public class KafkaStreamsStateStoreIntegrationTests { Thread.sleep(1000); // assertions - ProductCountApplication productCount = context - .getBean(ProductCountApplication.class); - WindowStore state = productCount.state; - assertThat(state != null).isTrue(); - assertThat(state.name()).isEqualTo("mystate"); - assertThat(state.persistent()).isTrue(); - assertThat(productCount.processed).isTrue(); + if (clazz.isAssignableFrom(ProductCountApplication.class)) { + ProductCountApplication productCount = context + .getBean(ProductCountApplication.class); + WindowStore state = productCount.state; + assertThat(state != null).isTrue(); + assertThat(state.name()).isEqualTo("mystate"); + assertThat(state.persistent()).isTrue(); + assertThat(productCount.processed).isTrue(); + } + else if (clazz.isAssignableFrom(StateStoreBeanApplication.class)) { + StateStoreBeanApplication productCount = context + .getBean(StateStoreBeanApplication.class); + WindowStore state = productCount.state; + assertThat(state != null).isTrue(); + assertThat(state.name()).isEqualTo("mystate"); + assertThat(state.persistent()).isTrue(); + assertThat(productCount.processed).isTrue(); + } + else { + fail("Expected assertiond did not happen"); + } + } @EnableBinding(KafkaStreamsProcessorX.class) @@ -170,6 +220,49 @@ public class KafkaStreamsStateStoreIntegrationTests { } } + @EnableBinding(KafkaStreamsProcessorZ.class) + @EnableAutoConfiguration + public static class StateStoreBeanApplication { + + WindowStore state; + + boolean processed; + + @StreamListener("input3") + @SuppressWarnings({"unchecked" }) + public void process(KStream input) { + + input.process(() -> new Processor() { + + @Override + public void init(ProcessorContext processorContext) { + state = (WindowStore) processorContext.getStateStore("mystate"); + } + + @Override + public void process(Object s, Product product) { + processed = true; + } + + @Override + public void close() { + if (state != null) { + state.close(); + } + } + }, "mystate"); + } + + @Bean + public StoreBuilder mystore() { + return Stores.windowStoreBuilder( + Stores.persistentWindowStore("mystate", + 3L, 3, 3L, false), Serdes.String(), + Serdes.String()); + } + } + + @EnableBinding(KafkaStreamsProcessorY.class) @EnableAutoConfiguration public static class ProductCountApplicationWithMultipleInputBindings { @@ -236,4 +329,10 @@ public class KafkaStreamsStateStoreIntegrationTests { @Input("input2") KStream input2(); } + + interface KafkaStreamsProcessorZ { + + @Input("input3") + KStream input3(); + } }