diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java
index 71cf073fd..0e7e99785 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java
@@ -19,9 +19,11 @@ package org.springframework.cloud.stream.binder.kafka.streams.function;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 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.TreeMap;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
@@ -107,6 +109,9 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
 		final String definition = streamFunctionProperties.getDefinition();
 		final String[] functionUnits = StringUtils.hasText(definition) ? definition.split(";") : new String[]{};
 
+		final Set<String> kafkaStreamsMethodNames = new HashSet<>(kafkaStreamsOnlyResolvableTypes.keySet());
+		kafkaStreamsMethodNames.addAll(this.resolvableTypeMap.keySet());
+
 		if (functionUnits.length == 0) {
 			for (String s : getResolvableTypes().keySet()) {
 				ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(s)};
@@ -123,21 +128,30 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
 					ResolvableType[] resolvableTypes = new ResolvableType[composedFunctions.length];
 
 					int i = 0;
+					boolean nonKafkaStreamsFunctionsFound = false;
+
 					for (String split : composedFunctions) {
 						derivedNameFromComposed = derivedNameFromComposed.concat(split);
 						resolvableTypes[i++] = getResolvableTypes().get(split);
+						if (!kafkaStreamsMethodNames.contains(split)) {
+							nonKafkaStreamsFunctionsFound = true;
+							break;
+						}
+					}
+					if (!nonKafkaStreamsFunctionsFound) {
+						RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
+								KafkaStreamsBindableProxyFactory.class);
+						registerKakaStreamsProxyFactory(registry, derivedNameFromComposed, resolvableTypes, rootBeanDefinition);
 					}
-
-					RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
-							KafkaStreamsBindableProxyFactory.class);
-					registerKakaStreamsProxyFactory(registry, derivedNameFromComposed, resolvableTypes, rootBeanDefinition);
 				}
 				else {
-
-					ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(functionUnit)};
-					RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
-							KafkaStreamsBindableProxyFactory.class);
-					registerKakaStreamsProxyFactory(registry, functionUnit, resolvableTypes, rootBeanDefinition);
+					// Ensure that the function unit is a Kafka Streams function
+					if (kafkaStreamsMethodNames.contains(functionUnit)) {
+						ResolvableType[] resolvableTypes = new ResolvableType[]{getResolvableTypes().get(functionUnit)};
+						RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
+								KafkaStreamsBindableProxyFactory.class);
+						registerKakaStreamsProxyFactory(registry, functionUnit, resolvableTypes, rootBeanDefinition);
+					}
 				}
 			}
 		}
diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java
index acc2d9173..aa26f82e8 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java
@@ -73,15 +73,16 @@ public class KafkaStreamsFunctionProcessorInvoker {
 				}
 				Optional<KafkaStreamsBindableProxyFactory> proxyFactory =
 						Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(derivedNameFromComposed[0])).findFirst();
-
-				this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(composedFunctions[0]),
-						derivedNameFromComposed[0], proxyFactory.get(), methods.get(derivedNameFromComposed[0]), resolvableTypeMap.get(composedFunctions[composedFunctions.length - 1]), composedFunctions);
+				proxyFactory.ifPresent(kafkaStreamsBindableProxyFactory ->
+						this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(composedFunctions[0]),
+						derivedNameFromComposed[0], kafkaStreamsBindableProxyFactory, methods.get(derivedNameFromComposed[0]), resolvableTypeMap.get(composedFunctions[composedFunctions.length - 1]), composedFunctions));
 			}
 			else {
 				Optional<KafkaStreamsBindableProxyFactory> proxyFactory =
 						Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(functionUnit)).findFirst();
-				this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(functionUnit), functionUnit,
-						proxyFactory.get(), methods.get(functionUnit), null);
+				proxyFactory.ifPresent(kafkaStreamsBindableProxyFactory ->
+						this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(functionUnit), functionUnit,
+						kafkaStreamsBindableProxyFactory, methods.get(functionUnit), null));
 			}
 		}
 	}