Honor Emission result in FluxMessageChannel

* Implement `Emission` handling in the `FluxMessageChannel` and `IntegrationReactiveUtils`
* Upgrade to Spring Kafka `2.6.0`
* Fix R2DBC components for deprecation in the Spring Data R2DBC
* Implement `StatementMapper.SelectSpec` for query expression
* Clean up for some sporadic test failures

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2015-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,9 @@
 
 package org.springframework.integration.channel;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
@@ -50,6 +53,8 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	private final Disposable.Composite upstreamSubscriptions = Disposables.composite();
 
+	private volatile boolean active = true;
+
 	public FluxMessageChannel() {
 		this.sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
 		this.processor = FluxProcessor.fromSink(this.sink);
@@ -57,9 +62,34 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
-		Assert.state(this.processor.hasDownstreams(),
+		Assert.state(this.active && this.processor.hasDownstreams(),
 				() -> "The [" + this + "] doesn't have subscribers to accept messages");
-		return this.sink.tryEmitNext(message).hasSucceeded();
+		long remainingTime = 0;
+		if (timeout > 0) {
+			remainingTime = timeout;
+		}
+		long parkTimeout = 10; // NOSONAR
+		long parkTimeoutNs = TimeUnit.MILLISECONDS.toNanos(parkTimeout);
+		while (this.active && !tryEmitMessage(message)) {
+			if (timeout >= 0 && (remainingTime -= parkTimeout) <= 0) {
+				return false;
+			}
+			LockSupport.parkNanos(parkTimeoutNs);
+		}
+		return true;
+	}
+
+	private boolean tryEmitMessage(Message<?> message) {
+		switch (this.sink.tryEmitNext(message)) {
+			case FAIL_OVERFLOW:
+				return false;
+			case FAIL_TERMINATED:
+			case FAIL_CANCELLED:
+				throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: "
+						+ this.sink);
+			default:
+				return true;
+		}
 	}
 
 	@Override
@@ -89,6 +119,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
 
 	@Override
 	public void destroy() {
+		this.active = false;
 		this.subscribedSignal.emitNext(false);
 		this.upstreamSubscriptions.dispose();
 		this.processor.onComplete();

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

@@ -126,11 +126,22 @@ public final class IntegrationReactiveUtils {
 
 	private static <T> Flux<Message<T>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
 		return Flux.defer(() -> {
-			Sinks.Many<Message<T>> sink = Sinks.many().multicast().onBackpressureBuffer(1);
-			@SuppressWarnings("unchecked")
+			Sinks.Many<Message<T>> sink = Sinks.many().unicast().onBackpressureError();
 			MessageHandler messageHandler = (message) -> {
-				while (!sink.tryEmitNext((Message<T>) message).hasSucceeded()) {
-					LockSupport.parkNanos(100); // NOSONAR
+				while (true) {
+					@SuppressWarnings("unchecked")
+					Sinks.Emission emission = sink.tryEmitNext((Message<T>) message);
+					switch (emission) {
+						case FAIL_OVERFLOW:
+							LockSupport.parkNanos(1000); // NOSONAR
+							break;
+						case FAIL_TERMINATED:
+						case FAIL_CANCELLED:
+							throw new IllegalStateException("Cannot emit messages into the cancelled " +
+									"or terminated sink for message channel: " + inputChannel);
+						default:
+							return;
+					}
 				}
 			};
 			inputChannel.subscribe(messageHandler);

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java

@@ -51,7 +51,6 @@ import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.MethodInvokingMessageHandler;
-import org.springframework.integration.test.condition.LogLevels;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
@@ -175,7 +174,6 @@ public class ReactiveStreamsConsumerTests {
 		reactiveConsumer.stop();
 	}
 
-	@LogLevels(level = "trace", categories = "org.springframework.integration")
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testReactiveStreamsConsumerPollableChannel() throws InterruptedException {