From 0f2285d5a3896afe1f8a72162070710e9545d8c4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 10 Sep 2021 17:23:01 -0400 Subject: [PATCH] Fix reply producing to not block reactive thread When `DirectChannel` is used for reply producing, the data is handled on the same thread which has produced it (normally), so if we have a request-reply afterwards (e.g. `gateway()`), this thread is blocked waiting for reply. When the thread is assumed to be non-blocked (e.g. Netty event loop), the request-reply withing such a thread for the same non-blocking client causes a deadlock: the thread waits for reply, but at the same time it supposes to fulfil a synchronization barrier with that reply * Fix `AbstractMessageProducingHandler.asyncNonReactiveReply()` to use a `publishOn(Schedulers.boundedElastic())` for reply `Mono` to free producing thread from potential downstream blocking * Demonstrate deadlock with a new test in the `RSocketDslTests`; the original report was against WebFlux, but conditions are really the same: `reactor-netty` is used as a low-level client **Cherry-pick to `5.4.x`** --- .../AbstractMessageProducingHandler.java | 5 +- .../rsocket/dsl/RSocketDslTests.java | 48 ++++++++++++++++++- 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 03c1d87394..ebc1da0c6f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -55,6 +55,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; /** * The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}. @@ -362,7 +363,9 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan else { reactiveReply = Mono.from((Publisher) reply); } - reactiveReply.subscribe(settableListenableFuture::set, settableListenableFuture::setException); + reactiveReply + .publishOn(Schedulers.boundedElastic()) + .subscribe(settableListenableFuture::set, settableListenableFuture::setException); future = settableListenableFuture; } future.addCallback(new ReplyFutureCallback(requestMessage, replyChannel)); diff --git a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java index f198d21f33..99c1713c90 100644 --- a/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java +++ b/spring-integration-rsocket/src/test/java/org/springframework/integration/rsocket/dsl/RSocketDslTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-2021 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,8 @@ package org.springframework.integration.rsocket.dsl; +import static org.assertj.core.api.Assertions.assertThat; + import java.time.Duration; import java.util.function.Function; @@ -28,6 +30,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.context.IntegrationFlowContext; import org.springframework.integration.rsocket.ClientRSocketConnector; import org.springframework.integration.rsocket.RSocketInteractionModel; import org.springframework.integration.rsocket.ServerRSocketConnector; @@ -74,6 +77,39 @@ public class RSocketDslTests { .verifyComplete(); } + @Autowired + IntegrationFlowContext integrationFlowContext; + + @Autowired + ClientRSocketConnector clientRSocketConnector; + + @Test + void testNoBlockingForReactiveThreads() { + IntegrationFlow flow = + f -> f + .handle(RSockets.outboundGateway("/lowercase") + .clientRSocketConnector(this.clientRSocketConnector)) + .transform("{ firstResult: payload }") + .enrich(e -> e + .requestPayloadExpression("payload.firstResult") + .requestSubFlow( + sf -> sf + .handle(RSockets.outboundGateway("/lowercase") + .clientRSocketConnector(this.clientRSocketConnector))) + .propertyExpression("secondResult", "payload")) + .transform("payload.values().toString()"); + + IntegrationFlowContext.IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(flow).register(); + + String result = flowRegistration.getMessagingTemplate().convertSendAndReceive("TEST", String.class); + + assertThat(result).isEqualTo("[test, test]"); + + flowRegistration.destroy(); + } + + @Configuration @EnableIntegration public static class TestConfiguration { @@ -96,7 +132,7 @@ public class RSocketDslTests { return IntegrationFlows .from(Function.class) .handle(RSockets.outboundGateway(message -> - message.getHeaders().getOrDefault("route", "/uppercase")) + message.getHeaders().getOrDefault("route", "/uppercase")) .interactionModel((message) -> RSocketInteractionModel.requestChannel) .expectedResponseType("T(java.lang.String)") .clientRSocketConnector(clientRSocketConnector), @@ -126,6 +162,14 @@ public class RSocketDslTests { .get(); } + @Bean + public IntegrationFlow rsocketLowerCaseFlow() { + return IntegrationFlows + .from(RSockets.inboundGateway("/lowercase")) + ., Flux>transform((flux) -> flux.map(String::toLowerCase)) + .get(); + } + } }