From 4247951d37d0701ecad78921ad4493fd32c3893a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 30 Sep 2022 16:50:56 -0400 Subject: [PATCH 1/6] GH-3635: Add Future & Mono to gateway Fixes https://github.com/spring-projects/spring-integration/issues/3635 When `Future` & `Mono` is used as a messaging gateway return type, the application hangs out on this barrier which may lead to the out of memory eventually * Add support for the `Future` & `Mono` messaging gateway return type and ensure an asynchronous call for the `gateway.send(Message)` operation and its exception handling. In case of successful call, the `Future` is fulfilled with `null` and `Mono` is completed as empty --- .../gateway/GatewayProxyFactoryBean.java | 9 +- .../gateway/AsyncGatewayTests.java | 83 ++++++++++++++----- src/reference/asciidoc/gateway.adoc | 13 +-- src/reference/asciidoc/whats-new.adoc | 4 +- 4 files changed, 81 insertions(+), 28 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 9631d56c1bf..0adeb94dbc0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -564,7 +564,7 @@ private Object invokeGatewayMethod(MethodInvocation invocation, boolean runningO } boolean shouldReturnMessage = Message.class.isAssignableFrom(gateway.returnType) || (!runningOnCallerThread && gateway.expectMessage); - boolean shouldReply = gateway.returnType != void.class; + boolean shouldReply = !gateway.isVoidReturn; int paramCount = method.getParameterTypes().length; Object response; boolean hasPayloadExpression = findPayloadExpression(method); @@ -640,7 +640,12 @@ private Object sendOrSendAndReceive(MethodInvocation invocation, MethodInvocatio } } else { - gateway.send(args); + if (gateway.isMonoReturn) { + return Mono.fromRunnable(() -> gateway.send(args)); + } + else { + gateway.send(args); + } } return null; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index d93dfb9c8b9..d2873728f7e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -30,10 +30,13 @@ import org.junit.jupiter.api.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -42,6 +45,7 @@ import org.springframework.messaging.support.MessageBuilder; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * @author Mark Fisher @@ -70,7 +74,7 @@ public void futureWithMessageReturned() throws Exception { } @Test - public void futureWithError() throws Exception { + public void futureWithError() { final Error error = new Error("error"); DirectChannel channel = new DirectChannel() { @@ -140,7 +144,7 @@ public void customFutureReturned() { } @Test - public void nonAsyncFutureReturned() throws Exception { + public void nonAsyncFutureReturned() { QueueChannel requestChannel = new QueueChannel(); addThreadEnricher(requestChannel); startResponder(requestChannel); @@ -204,6 +208,30 @@ public void futureWithWildcardReturned() throws Exception { assertThat(result).isEqualTo("foobar"); } + @Test + public void futureVoid() throws Exception { + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class); + proxyFactory.setDefaultRequestChannel(new NullChannel()); + proxyFactory.setBeanName("testGateway"); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Future f = service.asyncSendAndForget("test1"); + Object result = f.get(10, TimeUnit.SECONDS); + assertThat(result).isNull(); + + new DirectFieldAccessor(proxyFactory).setPropertyValue("initialized", false); + proxyFactory.setDefaultRequestChannel((message, timeout) -> { + throw new MessageDispatchingException(message, "intentional dispatcher error"); + }); + proxyFactory.afterPropertiesSet(); + + Future futureError = service.asyncSendAndForget("test2"); + assertThatExceptionOfType(ExecutionException.class) + .isThrownBy(() -> futureError.get(10, TimeUnit.SECONDS)) + .withCauseInstanceOf(MessageDispatchingException.class) + .withMessageContaining("intentional dispatcher error"); + } @Test public void monoWithMessageReturned() { @@ -252,7 +280,7 @@ public void monoWithWildcardReturned() { } @Test - public void monoWithConsumer() throws Exception { + public void monoWithConsumer() { QueueChannel requestChannel = new QueueChannel(); startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class); @@ -263,16 +291,38 @@ public void monoWithConsumer() throws Exception { TestEchoService service = (TestEchoService) proxyFactory.getObject(); Mono mono = service.returnStringPromise("foo"); - final AtomicReference result = new AtomicReference<>(); - final CountDownLatch latch = new CountDownLatch(1); + StepVerifier.create(mono) + .expectNext("foobar") + .verifyComplete(); + } - mono.subscribe(s -> { - result.set(s); - latch.countDown(); + @Test + public void monoVoid() throws InterruptedException { + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class); + proxyFactory.setDefaultRequestChannel(new NullChannel()); + proxyFactory.setBeanName("testGateway"); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Mono mono = service.monoVoid("test1"); + + CountDownLatch emptyMonoLatch = new CountDownLatch(1); + mono.switchIfEmpty(Mono.empty().doOnSuccess(v -> emptyMonoLatch.countDown()).then()).subscribe(); + + assertThat(emptyMonoLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + new DirectFieldAccessor(proxyFactory).setPropertyValue("initialized", false); + proxyFactory.setDefaultRequestChannel((message, timeout) -> { + throw new MessageDispatchingException(message, "intentional dispatcher error"); }); + proxyFactory.afterPropertiesSet(); - latch.await(10, TimeUnit.SECONDS); - assertThat(result.get()).isEqualTo("foobar"); + Mono monoError = service.monoVoid("test2"); + + StepVerifier.create(monoError) + .expectSubscription() + .expectError(MessageDispatchingException.class) + .verify(Duration.ofSeconds(10)); } private static void startResponder(final PollableChannel requestChannel) { @@ -323,18 +373,13 @@ private interface TestEchoService { Mono returnSomethingPromise(String s); - } + Future asyncSendAndForget(String s); - private static class CustomFuture implements Future { + Mono monoVoid(String s); - private final String result; - - private final Thread thread; + } - private CustomFuture(String result, Thread thread) { - this.result = result; - this.thread = thread; - } + private record CustomFuture(String result, Thread thread) implements Future { @Override public boolean cancel(boolean mayInterruptIfRunning) { diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index 5856707d129..bb74f25b5d5 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -460,7 +460,7 @@ If you provide a one-way flow, nothing would be sent back to the caller. If you want to completely suppress exceptions, you can provide a reference to the global `nullChannel` (essentially a `/dev/null` approach). Finally, as mentioned above, if no `error-channel` is defined, then the exceptions propagate as usual. -When you use the `@MessagingGateway` annotation (see `<>`), you can use use the `errorChannel` attribute. +When you use the `@MessagingGateway` annotation (see `<>`), you can use an `errorChannel` attribute. Starting with version 5.0, when you use a gateway method with a `void` return type (one-way flow), the `error-channel` reference (if provided) is populated in the standard `errorChannel` header of each sent message. This feature allows a downstream asynchronous flow, based on the standard `ExecutorChannel` configuration (or a `QueueChannel`), to override a default global `errorChannel` exceptions sending behavior. @@ -469,7 +469,7 @@ The `error-channel` property was ignored for `void` methods with an asynchronous Instead, error messages were sent to the default `errorChannel`. -IMPORTANT: Exposing the messaging system through simple POJI Gateways provides benefits, but "`hiding`" the reality of the underlying messaging system does come at a price, so there are certain things you should consider. +IMPORTANT: Exposing the messaging system through simple POJI Gateways provides benefits, but "`hiding`" the reality of the underlying messaging system does come at a price, so there are certain things you should consider. We want our Java method to return as quickly as possible and not hang for an indefinite amount of time while the caller is waiting on it to return (whether void, a return value, or a thrown Exception). When regular methods are used as a proxies in front of the messaging system, we have to take into account the potentially asynchronous nature of the underlying messaging. This means that there might be a chance that a message that was initiated by a gateway could be dropped by a filter and never reach a component that is responsible for producing a reply. @@ -782,10 +782,9 @@ The calling thread continues, with `handleInvoice()` being called when the flow As mentioned in the <> section above, if you wish some downstream component to return a message with an async payload (`Future`, `Mono`, and others), you must explicitly set the async executor to `null` (or `""` when using XML configuration). The flow is then invoked on the caller thread and the result can be retrieved later. -===== `void` Return Type +===== Asynchronous `void` Return Type -Unlike the return types mentioned earlier, when the method return type is `void`, the framework cannot implicitly determine that you wish the downstream flow to run asynchronously, with the caller thread returning immediately. -In this case, you must annotate the interface method with `@Async`, as the following example shows: +The messaging gateway method can be declared like this: ==== [source, java] @@ -801,7 +800,9 @@ public interface MyGateway { ---- ==== -Unlike the `Future` return types, there is no way to inform the caller if some exception is thrown by the flow, unless some custom `TaskExecutor` (such as an `ErrorHandlingTaskExecutor`) is associated with the `@Async` annotation. +But downstream exceptions are not going to be propagated back to the caller. +To ensure asynchronous behaviour for downstream flow invocation and exception propagation to the caller, starting with version 6.0, the framework provides support for the `Future` and `Mono` return types. +The use-case is similar to send-and-forget behavior described before for plain `void` return type, but with a difference that flow execution happens asynchronously and returned `Future` (or `Mono`) is complete with a `null` or exceptionally according to the `send` operation result. [[gateway-no-response]] ==== Gateway Behavior When No response Arrives diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 8b061d1b51b..c03c99dc370 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -80,7 +80,9 @@ See <<./dsl.adoc#java-dsl,Java DSL>> for more information. The `org.springframework.util.concurrent.ListenableFuture` has been deprecated starting with Spring Framework `6.0`. All Spring Integration async API has been migrated to the `CompletableFuture`. -See <<./gateway.adoc#gw-completable-future, CompletableFuture support>> for more information. +Also Messaging Gateway interface method can now return `Future` and `Mono` with a proper asynchronous execution of the downstream flow. + +See <<./gateway.adoc#async-gateway, Asynchronous Gateway>> for more information. The `integrationGlobalProperties` bean is now declared by the framework as an instance of `org.springframework.integration.context.IntegrationProperties` instead of the previously deprecated `java.util.Properties`. From 00082b38d1d938d2151aba90f9eddbcba5983bbd Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sat, 1 Oct 2022 11:58:39 -0400 Subject: [PATCH 2/6] * Check for `void.class` as well in the `GatewayProxyFactoryBean.isVoidReturnType` --- .../integration/gateway/GatewayProxyFactoryBean.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 0adeb94dbc0..24eef0e455c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -1099,7 +1099,7 @@ private boolean isVoidReturnType(ResolvableType resolvableType) { if (Future.class.isAssignableFrom(this.returnType) || Mono.class.isAssignableFrom(this.returnType)) { returnTypeToCheck = resolvableType.getGeneric(0).resolve(Object.class); } - return Void.class.isAssignableFrom(returnTypeToCheck); + return Void.class.isAssignableFrom(returnTypeToCheck) || void.class.isAssignableFrom(returnTypeToCheck); } private void setPollable() { From 9ebec2c709d17d335458b5ba16be9bcfaa3ce634 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 3 Oct 2022 17:38:50 -0400 Subject: [PATCH 3/6] * Allow `Future` as a reply type of the gateway request-reply operation --- .../gateway/GatewayProxyFactoryBean.java | 9 ++--- .../gateway/AsyncGatewayTests.java | 33 +++++++++++++++++++ src/reference/asciidoc/gateway.adoc | 8 ++++- 3 files changed, 45 insertions(+), 5 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 24eef0e455c..dc52c03c324 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -564,15 +564,16 @@ private Object invokeGatewayMethod(MethodInvocation invocation, boolean runningO } boolean shouldReturnMessage = Message.class.isAssignableFrom(gateway.returnType) || (!runningOnCallerThread && gateway.expectMessage); - boolean shouldReply = !gateway.isVoidReturn; + boolean oneWay = + void.class.isAssignableFrom(gateway.returnType) || (gateway.isVoidReturn && !runningOnCallerThread); int paramCount = method.getParameterTypes().length; Object response; boolean hasPayloadExpression = findPayloadExpression(method); if (paramCount == 0 && !hasPayloadExpression) { - response = receive(gateway, method, shouldReply, shouldReturnMessage); + response = receive(gateway, method, !oneWay, shouldReturnMessage); } else { - response = sendOrSendAndReceive(invocation, gateway, shouldReturnMessage, shouldReply); + response = sendOrSendAndReceive(invocation, gateway, shouldReturnMessage, !oneWay); } return response(gateway.returnType, shouldReturnMessage, response); } @@ -1099,7 +1100,7 @@ private boolean isVoidReturnType(ResolvableType resolvableType) { if (Future.class.isAssignableFrom(this.returnType) || Mono.class.isAssignableFrom(this.returnType)) { returnTypeToCheck = resolvableType.getGeneric(0).resolve(Object.class); } - return Void.class.isAssignableFrom(returnTypeToCheck) || void.class.isAssignableFrom(returnTypeToCheck); + return Void.class.isAssignableFrom(returnTypeToCheck); } private void setPollable() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index d2873728f7e..24f03f10e3d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -33,8 +33,10 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.MessageDispatchingException; +import org.springframework.integration.annotation.AnnotationConstants; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; +import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; @@ -42,6 +44,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import reactor.core.publisher.Mono; @@ -233,6 +236,35 @@ public void futureVoid() throws Exception { .withMessageContaining("intentional dispatcher error"); } + @Test + public void futureVoidReply() throws Exception { + QueueChannel requestChannel = new QueueChannel(); + CountDownLatch readyForReplyLatch = new CountDownLatch(1); + new Thread(() -> { + try { + Message input = requestChannel.receive(); + CompletableFuture reply = new CompletableFuture<>(); + ((MessageChannel) input.getHeaders().getReplyChannel()).send(new GenericMessage<>(reply)); + readyForReplyLatch.await(10, TimeUnit.SECONDS); + reply.complete(null); + } + catch (InterruptedException e) { + System.err.println(e); + } + }).start(); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class); + proxyFactory.setDefaultRequestChannel(requestChannel); + proxyFactory.setBeanName("testGateway"); + proxyFactory.setBeanFactory(mock(BeanFactory.class)); + proxyFactory.setAsyncExecutor(null); + proxyFactory.afterPropertiesSet(); + TestEchoService service = (TestEchoService) proxyFactory.getObject(); + Future f = service.sendAndReceiveFutureVoid("test"); + readyForReplyLatch.countDown(); + Object result = f.get(10, TimeUnit.SECONDS); + assertThat(result).isNull(); + } + @Test public void monoWithMessageReturned() { QueueChannel requestChannel = new QueueChannel(); @@ -375,6 +407,7 @@ private interface TestEchoService { Future asyncSendAndForget(String s); + Future sendAndReceiveFutureVoid(String s); Mono monoVoid(String s); } diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index bb74f25b5d5..10d4c9ecf7b 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -801,9 +801,15 @@ public interface MyGateway { ==== But downstream exceptions are not going to be propagated back to the caller. -To ensure asynchronous behaviour for downstream flow invocation and exception propagation to the caller, starting with version 6.0, the framework provides support for the `Future` and `Mono` return types. +To ensure asynchronous behavior for downstream flow invocation and exception propagation to the caller, starting with version 6.0, the framework provides support for the `Future` and `Mono` return types. The use-case is similar to send-and-forget behavior described before for plain `void` return type, but with a difference that flow execution happens asynchronously and returned `Future` (or `Mono`) is complete with a `null` or exceptionally according to the `send` operation result. +NOTE: If the `Future` is exact downstream flow reply, then an `asyncExecutor` option of the gateway must be set to null (`AnnotationConstants.NULL` for a `@MessagingGateway` configuration) and the `send` part is performed on a producer thread. +The reply one depends on the downstream flow configuration. +This way it is up target application to produce a `Future` reply correctly. +The `Mono` use-case is already out of the framework threading control, so setting `asyncExecutor` to null won't make sense. +There `Mono` as a result of the request-reply gateway operation must be configured as a `Mono` return type of the gateway method. + [[gateway-no-response]] ==== Gateway Behavior When No response Arrives From 5e7454e8c22bd5c839e0e845381ddf2577efd6d8 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 3 Oct 2022 17:56:50 -0400 Subject: [PATCH 4/6] * Fix Checkstyle violations --- .../springframework/integration/gateway/AsyncGatewayTests.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 24f03f10e3d..6070df157b4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -33,10 +33,8 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.MessageDispatchingException; -import org.springframework.integration.annotation.AnnotationConstants; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; -import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; @@ -408,6 +406,7 @@ private interface TestEchoService { Future asyncSendAndForget(String s); Future sendAndReceiveFutureVoid(String s); + Mono monoVoid(String s); } From 3758a95826b05038a9272186af8cf8d9d828c88b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 3 Oct 2022 17:59:28 -0400 Subject: [PATCH 5/6] * Resolve `System.err.println()` in the test code --- .../integration/gateway/AsyncGatewayTests.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 6070df157b4..6f396731031 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -44,6 +44,7 @@ import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.ReflectionUtils; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -246,8 +247,8 @@ public void futureVoidReply() throws Exception { readyForReplyLatch.await(10, TimeUnit.SECONDS); reply.complete(null); } - catch (InterruptedException e) { - System.err.println(e); + catch (InterruptedException ex) { + ReflectionUtils.rethrowRuntimeException(ex); } }).start(); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(TestEchoService.class); From 54f08a1350a9fb1192657d632fee15e07b3ec347 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 4 Oct 2022 15:58:56 -0400 Subject: [PATCH 6/6] Add `Thread.currentThread.interrupt()` to the `InterruptedException` block in the test --- .../springframework/integration/gateway/AsyncGatewayTests.java | 1 + 1 file changed, 1 insertion(+) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 6f396731031..fac9f9ee2b7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -248,6 +248,7 @@ public void futureVoidReply() throws Exception { reply.complete(null); } catch (InterruptedException ex) { + Thread.currentThread.interrupt(); ReflectionUtils.rethrowRuntimeException(ex); } }).start();