diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
index 5a791ec654..1c81fe2d5c 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java
@@ -172,16 +172,17 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 		else if (this.subscriber != null) {
 			this.subscription =
 					Flux.from(this.publisher)
-							.doOnComplete(this.subscriber::onComplete)
-							.doOnSubscribe(this.subscriber::onSubscribe)
 							.subscribe((data) -> {
-								try {
-									this.subscriber.onNext(data);
-								}
-								catch (Exception ex) {
-									this.errorHandler.handleError(ex);
-								}
-							});
+										try {
+											this.subscriber.onNext(data);
+										}
+										catch (Exception ex) {
+											this.errorHandler.handleError(ex);
+										}
+									},
+									null,
+									this.subscriber::onComplete,
+									this.subscriber::onSubscribe);
 		}
 	}