diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
index ff670fc0f3..70cb26ad2f 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
@@ -19,6 +19,8 @@ package org.springframework.integration.util;
 import java.time.Duration;
 import java.util.concurrent.locks.LockSupport;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
 
 import org.springframework.integration.StaticMessageHeaderAccessor;
@@ -45,6 +47,8 @@ import reactor.core.scheduler.Schedulers;
  */
 public final class IntegrationReactiveUtils {
 
+	private static final Log logger = LogFactory.getLog(IntegrationReactiveUtils.class);
+
 	/**
 	 * The subscriber context entry for {@link Flux#delayElements}
 	 * from the {@link Mono#repeatWhenEmpty(java.util.function.Function)}.
@@ -76,14 +80,18 @@ public final class IntegrationReactiveUtils {
 		return Mono.
 				<Message<T>>create(monoSink ->
 						monoSink.onRequest(value -> monoSink.success(messageSource.receive())))
-				.doOnSuccess((message) ->
-						AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)))
+				.doOnSuccess((message) -> {
+					if (message != null) {
+						AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message));
+					}
+				})
 				.doOnError(MessagingException.class,
 						(ex) -> {
 							Message<?> failedMessage = ex.getFailedMessage();
 							if (failedMessage != null) {
 								AckUtils.autoNack(StaticMessageHeaderAccessor.getAcknowledgmentCallback(failedMessage));
 							}
+							logger.error("Error from Flux for : " + messageSource, ex);
 						})
 				.subscribeOn(Schedulers.boundedElastic())
 				.repeatWhenEmpty((repeat) ->