State store beans registered with StreamListener

When StreamListener based processors used in Kafka Streams
applications, it is not possible to register state store beans
using StateStoreBuilder. This is allowed in the functional model.
This method of providing custom state stores is desired as this
gives the user more flexibility in configuring Serde's and other
properties on the state store. Adding this feature to StreamListner
based Kafka Streams processors.

Adding test to verify.

Adding docs.

Resolves #676

docs/src/main/asciidoc/kafka-streams.adoc

@@ -942,7 +942,22 @@ public KStream<?, WordCount> process(KStream<Object, String> 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.

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<String, StoreBuilder> 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);
+	}
 }

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<String, StoreBuilder> 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

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,

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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
@@ -126,13 +161,28 @@ public class KafkaStreamsStateStoreIntegrationTests {
 		Thread.sleep(1000);
 
 		// assertions
-		ProductCountApplication productCount = context
-				.getBean(ProductCountApplication.class);
-		WindowStore<Object, String> 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<Object, String> 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<Object, String> 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<Object, String> state;
+
+		boolean processed;
+
+		@StreamListener("input3")
+		@SuppressWarnings({"unchecked" })
+		public void process(KStream<Object, Product> input) {
+
+			input.process(() -> new Processor<Object, Product>() {
+
+				@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();
+	}
 }