Optional io.micrometer:context-propagation (#8556)

For better performance by default it is better to not pull
a `io.micrometer:context-propagation` a hard dependency.

* Remove `io.micrometer:context-propagation` dependency management
* It is pulled transitively by the `io.micrometer:micrometer-tracing-integration-test` in test scope
* Rework all the `ContextSnapshot` usage in the reactive code to respective recommended `handle()` API in `Flux` and `Mono`

Author: artembilan

build.gradle

@@ -87,7 +87,6 @@ ext {
     lettuceVersion = '6.2.2.RELEASE'
     log4jVersion = '2.19.0'
     mailVersion = '1.0.0'
-    micrometerPropagationVersion = '1.0.2'
     micrometerTracingVersion = '1.0.2'
     micrometerVersion = '1.10.4'
     mockitoVersion = '4.10.0'
@@ -533,7 +532,6 @@ project('spring-integration-core') {
         }
         api 'io.projectreactor:reactor-core'
         api 'io.micrometer:micrometer-observation'
-        api "io.micrometer:context-propagation:$micrometerPropagationVersion"
 
         optionalApi 'com.fasterxml.jackson.core:jackson-databind'
         optionalApi 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'

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

@@ -20,7 +20,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
 
-import io.micrometer.context.ContextSnapshot;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import reactor.core.Disposable;
@@ -112,19 +111,18 @@ public void subscribeTo(Publisher<? extends Message<?>> publisher) {
 				Flux.from(publisher)
 						.delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next())
 						.publishOn(this.scheduler)
-						.transformDeferredContextual((flux, contextView) ->
-								flux.doOnNext((message) -> {
-									var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView);
-									try (scope) {
-										if (!send(message)) {
-											throw new MessageDeliveryException(message,
-													"Failed to send message to channel '" + this);
-										}
-									}
-									catch (Exception ex) {
-										logger.warn(ex, () -> "Error during processing event: " + message);
-									}
-								}))
+						.handle((message, synchronousSink) -> {
+							try {
+								if (!send(message)) {
+									logger.warn(new MessageDeliveryException(message,
+											"Failed to send message to channel '" + this),
+											"Message was not delivered");
+								}
+							}
+							catch (Exception ex) {
+								logger.warn(ex, () -> "Error during processing event: " + message);
+							}
+						})
 						.contextCapture()
 						.subscribe());
 	}

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -20,7 +20,6 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
-import io.micrometer.context.ContextSnapshot;
 import io.micrometer.observation.ObservationRegistry;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -721,7 +720,7 @@ private Mono<Message<?>> doSendAndReceiveMessageReactive(MessageChannel requestC
 			throw new MessageMappingException("Cannot map to message: " + object, e);
 		}
 
-		return Mono.deferContextual(contextView -> {
+		return Mono.defer(() -> {
 					Object originalReplyChannelHeader = requestMessage.getHeaders().getReplyChannel();
 					Object originalErrorChannelHeader = requestMessage.getHeaders().getErrorChannel();
 
@@ -739,13 +738,13 @@ private Mono<Message<?>> doSendAndReceiveMessageReactive(MessageChannel requestC
 							.setErrorChannel(replyChan)
 							.build();
 
-					var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView);
-					try (scope) {
-						sendMessageForReactiveFlow(requestChannel, messageToSend);
-					}
-
-					return buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error,
-							originalReplyChannelHeader, originalErrorChannelHeader);
+					return Mono.just(messageToSend)
+							.handle((message, synchronousSink) -> {
+								sendMessageForReactiveFlow(requestChannel, message);
+								synchronousSink.complete();
+							})
+							.then(buildReplyMono(requestMessage, replyChan.replyMono.asMono(), error,
+									originalReplyChannelHeader, originalErrorChannelHeader));
 				})
 				.onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t));
 	}

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java

@@ -27,7 +27,6 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
-import io.micrometer.context.ContextSnapshot;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -152,20 +151,20 @@ private Mono<Void> doHandle(ServerWebExchange exchange) {
 						new RequestEntity<>(body, exchange.getRequest().getHeaders(),
 								exchange.getRequest().getMethod(), exchange.getRequest().getURI()))
 				.flatMap(entity -> buildMessage(entity, exchange))
-				.flatMap(requestTuple ->
-						Mono.deferContextual(contextView -> {
-							if (isExpectReply()) {
-								return sendAndReceiveMessageReactive(requestTuple.getT1())
-										.flatMap(replyMessage -> populateResponse(exchange, replyMessage));
-							}
-							else {
-								var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView);
-								try (scope) {
-									send(requestTuple.getT1());
-								}
-								return setStatusCode(exchange, requestTuple.getT2());
-							}
-						}))
+				.flatMap(requestTuple -> {
+					if (isExpectReply()) {
+						return sendAndReceiveMessageReactive(requestTuple.getT1())
+								.flatMap(replyMessage -> populateResponse(exchange, replyMessage));
+					}
+					else {
+						return Mono.just(requestTuple.getT1())
+								.handle((objectMessage, synchronousSink) -> {
+									send(objectMessage);
+									synchronousSink.complete();
+								})
+								.then(setStatusCode(exchange, requestTuple.getT2()));
+					}
+				})
 				.doOnTerminate(this.activeCount::decrementAndGet);
 
 	}