Fix IntegrationReactiveUtils

The `Mono.doOnSuccess()` is always called for completed `MonoSink`
even if the value is `null`

* Fix `IntegrationReactiveUtils.messageSourceToFlux()` to check a
message for `null` before calling `AckUtils.autoAck()`
* Add an `logger.error()` message when `doOnError()` is invoked for that `Mono`

**Cherry-pick to 5.3.x**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java

@@ -19,6 +19,8 @@
 import java.time.Duration;
 import java.util.function.Function;
 
+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 @@
  */
 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(Function)}.
@@ -75,16 +79,19 @@ private IntegrationReactiveUtils() {
 	public static <T> Flux<Message<T>> messageSourceToFlux(MessageSource<T> messageSource) {
 		return Mono.
 				<Message<T>>create(monoSink ->
-						monoSink.onRequest(value ->
-								monoSink.success(messageSource.receive())))
-				.doOnSuccess((message) ->
-						AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)))
+						monoSink.onRequest(value -> monoSink.success(messageSource.receive())))
+				.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) ->