diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationChannel.java index bbaaddd293e..d4e01a572b1 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationChannel.java @@ -23,6 +23,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.remoting.exception.RemotingTimeoutException; import org.apache.rocketmq.remoting.protocol.RemotingCommand; public class InvocationChannel extends SimpleChannel { @@ -68,6 +69,7 @@ public void clearExpireContext() { Map.Entry entry = iterator.next(); if (entry.getValue().expired(ConfigurationManager.getProxyConfig().getChannelExpiredInSeconds())) { iterator.remove(); + entry.getValue().expire(new RemotingTimeoutException("Invocation context expired. opaque=" + entry.getKey())); count++; log.debug("An expired request is found, request: {}", entry.getValue()); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContext.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContext.java index 9fb488eb9b1..e3b610074f7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContext.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContext.java @@ -40,4 +40,9 @@ public CompletableFuture getResponse() { public void handle(RemotingCommand remotingCommand) { response.complete(remotingCommand); } + + @Override + public void expire(Throwable throwable) { + response.completeExceptionally(throwable); + } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContextInterface.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContextInterface.java index 0db9516486b..2c02bd4fc83 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContextInterface.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContextInterface.java @@ -23,4 +23,7 @@ public interface InvocationContextInterface { void handle(RemotingCommand remotingCommand); boolean expired(long expiredTimeSec); + + default void expire(Throwable throwable) { + } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/InvocationChannelTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/InvocationChannelTest.java index ddede4fbc86..0fb36c77761 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/InvocationChannelTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/InvocationChannelTest.java @@ -17,16 +17,24 @@ package org.apache.rocketmq.proxy.service.channel; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.junit.BeforeClass; import org.junit.Test; -import java.util.concurrent.atomic.AtomicBoolean; - import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; public class InvocationChannelTest { + @BeforeClass + public static void setUp() throws Exception { + ConfigurationManager.initEnv(); + ConfigurationManager.initConfig(); + } + @Test public void testWriteAndFlushShouldNotRemoveReRegisteredContext() { InvocationChannel channel = new InvocationChannel("127.0.0.1:8080", "127.0.0.1:8081"); @@ -64,4 +72,22 @@ public boolean expired(long expiredTimeSec) { assertTrue(nextContextHandled.get()); assertFalse(channel.isWritable()); } + + @Test + public void testClearExpireContextShouldCompleteResponseFutureExceptionally() { + int originalExpiredInSeconds = ConfigurationManager.getProxyConfig().getChannelExpiredInSeconds(); + ConfigurationManager.getProxyConfig().setChannelExpiredInSeconds(0); + try { + InvocationChannel channel = new InvocationChannel("127.0.0.1:8080", "127.0.0.1:8081"); + CompletableFuture responseFuture = new CompletableFuture<>(); + channel.registerInvocationContext(1, new InvocationContext(responseFuture)); + + channel.clearExpireContext(); + + assertFalse(channel.isWritable()); + assertTrue(responseFuture.isCompletedExceptionally()); + } finally { + ConfigurationManager.getProxyConfig().setChannelExpiredInSeconds(originalExpiredInSeconds); + } + } }