diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java
index 80806e92b..791457c24 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java
@@ -209,7 +209,8 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto
 			if (bean != null && (BiFunction.class.isAssignableFrom(bean.getClass()) || BiConsumer.class.isAssignableFrom(bean.getClass()))) {
 				numberOfInputs = 2;
 			}
-			else if (returnType.getRawClass().isAssignableFrom(Function.class) || returnType.getRawClass().isAssignableFrom(Consumer.class)) {
+			else if (returnType.getRawClass() != null && (returnType.getRawClass().isAssignableFrom(Function.class)
+				|| returnType.getRawClass().isAssignableFrom(Consumer.class))) {
 				numberOfInputs = 1;
 				ResolvableType arg1 = returnType;