diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
index 4cc0881c8..d9023c9b7 100644
--- a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
+++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
@@ -101,7 +101,7 @@ public class StreamConfiguration {
 		@ConditionalOnProperty("spring.cloud.stream.bindings.input.destination")
 		public StreamListeningConsumerInvoker<Object> invoker(FunctionCatalog registry) {
 			String name = properties.getEndpoint();
-			Consumer<Object> consumer = registry.lookupConsumer(name);
+			Consumer<Flux<Object>> consumer = registry.lookupConsumer(name);
 			return new StreamListeningConsumerInvoker<Object>(consumer);
 		}
 	}
diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamListeningConsumerInvoker.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamListeningConsumerInvoker.java
index 2ebaaa73c..e695eeb4b 100644
--- a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamListeningConsumerInvoker.java
+++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamListeningConsumerInvoker.java
@@ -29,14 +29,14 @@ import reactor.core.publisher.Flux;
  */
 public class StreamListeningConsumerInvoker<T> {
 
-	private final Consumer<T> consumer;
+	private final Consumer<Flux<T>> consumer;
 
-	public StreamListeningConsumerInvoker(Consumer<T> consumer) {
+	public StreamListeningConsumerInvoker(Consumer<Flux<T>> consumer) {
 		this.consumer = consumer;
 	}
 
 	@StreamListener
 	public void handle(@Input(Processor.INPUT) Flux<T> input) {
-		input.subscribe(this.consumer);
+		this.consumer.accept(input);
 	}
 }