[ISSUE #10764] Complete expired invocation futures

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<Integer, InvocationContextInterface> 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());
             }

proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/InvocationContext.java

@@ -40,4 +40,9 @@ public CompletableFuture<RemotingCommand> getResponse() {
     public void handle(RemotingCommand remotingCommand) {
         response.complete(remotingCommand);
     }
+
+    @Override
+    public void expire(Throwable throwable) {
+        response.completeExceptionally(throwable);
+    }
 }

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) {
+    }
 }

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<RemotingCommand> responseFuture = new CompletableFuture<>();
+            channel.registerInvocationContext(1, new InvocationContext(responseFuture));
+
+            channel.clearExpireContext();
+
+            assertFalse(channel.isWritable());
+            assertTrue(responseFuture.isCompletedExceptionally());
+        } finally {
+            ConfigurationManager.getProxyConfig().setChannelExpiredInSeconds(originalExpiredInSeconds);
+        }
+    }
 }