diff --git a/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java
index 166693de..5711987a 100644
--- a/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java
+++ b/functions/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java
@@ -63,7 +63,7 @@ public class AggregatorFunctionConfiguration {
 			FluxMessageChannel outputChannel) {
 
 		return input -> Flux.from(outputChannel)
-				.doOnSubscribe((sub) -> inputChannel.subscribeTo(input));
+				.doOnRequest((request) -> inputChannel.subscribeTo(input));
 	}
 
 	@Bean