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 853397c00..dc0b3931b 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
@@ -251,8 +251,10 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
 	}
 
 	protected KStream<?, ?> getKStream(String inboundName, BindingProperties bindingProperties, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties,
-									StreamsBuilder streamsBuilder, Serde<?> keySerde, Serde<?> valueSerde, Topology.AutoOffsetReset autoOffsetReset) {
-		addStateStoreBeans(streamsBuilder);
+									StreamsBuilder streamsBuilder, Serde<?> keySerde, Serde<?> valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) {
+		if (firstBuild) {
+			addStateStoreBeans(streamsBuilder);
+		}
 
 		String[] bindingTargets = StringUtils.commaDelimitedListToStringArray(
 				this.bindingServiceProperties.getBindingDestination(inboundName));
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 210bc7fc7..bea0e1a78 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
@@ -316,7 +316,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
 					final Topology.AutoOffsetReset autoOffsetReset = getAutoOffsetReset(input, extendedConsumerProperties);
 
 					if (parameterType.isAssignableFrom(KStream.class)) {
-						KStream<?, ?> stream = getKStream(input, bindingProperties, extendedConsumerProperties, streamsBuilder, keySerde, valueSerde, autoOffsetReset);
+						KStream<?, ?> stream = getKStream(input, bindingProperties, extendedConsumerProperties,
+								streamsBuilder, keySerde, valueSerde, autoOffsetReset, i == 0);
 						KStreamBoundElementFactory.KStreamWrapper kStreamWrapper =
 								(KStreamBoundElementFactory.KStreamWrapper) targetBean;
 						//wrap the proxy created during the initial target type binding with real object (KStream)
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 2a7d19622..505ca9341 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
@@ -276,7 +276,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr
 					if (parameterType.isAssignableFrom(KStream.class)) {
 						KStream<?, ?> stream = getkStream(inboundName, spec,
 								bindingProperties, extendedConsumerProperties, streamsBuilder, keySerde, valueSerde,
-								autoOffsetReset);
+								autoOffsetReset, parameterIndex == 0);
 						KStreamBoundElementFactory.KStreamWrapper kStreamWrapper = (KStreamBoundElementFactory.KStreamWrapper) targetBean;
 						// wrap the proxy created during the initial target type binding
 						// with real object (KStream)
@@ -369,7 +369,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr
 									BindingProperties bindingProperties,
 									KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, StreamsBuilder streamsBuilder,
 									Serde<?> keySerde, Serde<?> valueSerde,
-									Topology.AutoOffsetReset autoOffsetReset) {
+									Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) {
 		if (storeSpec != null) {
 			StoreBuilder storeBuilder = buildStateStore(storeSpec);
 			streamsBuilder.addStateStore(storeBuilder);
@@ -377,7 +377,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr
 				LOG.info("state store " + storeBuilder.name() + " added to topology");
 			}
 		}
-		return getKStream(inboundName, bindingProperties, kafkaStreamsConsumerProperties, streamsBuilder, keySerde, valueSerde, autoOffsetReset);
+		return getKStream(inboundName, bindingProperties, kafkaStreamsConsumerProperties, streamsBuilder,
+				keySerde, valueSerde, autoOffsetReset, firstBuild);
 	}
 
 	private void validateStreamListenerMethod(StreamListener streamListener,
diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java
index 4b91b2ad8..a564b196a 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java
@@ -105,9 +105,9 @@ public class KafkaStreamsFunctionStateStoreTests {
 		boolean processed;
 
 		@Bean
-		public java.util.function.Consumer<KStream<Object, String>> process() {
-			return input ->
-					input.process((ProcessorSupplier<Object, String>) () -> new Processor<Object, String>() {
+		public java.util.function.BiConsumer<KStream<Object, String>, KStream<Object, String>> process() {
+			return (input0, input1) ->
+					input0.process((ProcessorSupplier<Object, String>) () -> new Processor<Object, String>() {
 						@Override
 						@SuppressWarnings("unchecked")
 						public void init(ProcessorContext context) {