From a0f9c19196b2d5f1f5b2dbf9931799508dd85aad Mon Sep 17 00:00:00 2001 From: liuhy Date: Sun, 2 Aug 2026 22:50:23 -0700 Subject: [PATCH 1/2] [ISSUE #10764] Complete expired invocation futures --- .../service/channel/InvocationChannel.java | 2 ++ .../service/channel/InvocationContext.java | 5 ++++ .../channel/InvocationContextInterface.java | 3 ++ .../channel/InvocationChannelTest.java | 30 +++++++++++++++++-- 4 files changed, 38 insertions(+), 2 deletions(-) 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); + } + } } From 9b664776631e9030f88e4dd1df0be4d708c15b92 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 04:04:30 -0700 Subject: [PATCH 2/2] fix(proxy): enrich expired invocation diagnostics --- .../rocketmq/proxy/service/channel/InvocationChannel.java | 7 +++++-- .../proxy/service/channel/InvocationContextInterface.java | 5 +++++ .../proxy/service/channel/InvocationChannelTest.java | 6 ++++++ 3 files changed, 16 insertions(+), 2 deletions(-) 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 d4e01a572b1..3ac3a479bdb 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 @@ -65,11 +65,14 @@ public void eraseInvocationContext(int opaque) { public void clearExpireContext() { Iterator> iterator = inFlightRequestMap.entrySet().iterator(); int count = 0; + long expirationSeconds = ConfigurationManager.getProxyConfig().getChannelExpiredInSeconds(); while (iterator.hasNext()) { Map.Entry entry = iterator.next(); - if (entry.getValue().expired(ConfigurationManager.getProxyConfig().getChannelExpiredInSeconds())) { + if (entry.getValue().expired(expirationSeconds)) { iterator.remove(); - entry.getValue().expire(new RemotingTimeoutException("Invocation context expired. opaque=" + entry.getKey())); + entry.getValue().expire(new RemotingTimeoutException("Invocation context expired after " + + expirationSeconds + " seconds. opaque=" + entry.getKey() + ", remoteAddress=" + + remoteAddress + ", localAddress=" + localAddress)); count++; log.debug("An expired request is found, request: {}", entry.getValue()); } 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 2c02bd4fc83..e86b04c527b 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 @@ -24,6 +24,11 @@ public interface InvocationContextInterface { boolean expired(long expiredTimeSec); + /** + * Notifies a waiting invocation that its channel context has expired. + * Implementations that retain an asynchronous response future must override this method. + */ default void expire(Throwable throwable) { + // Stateless contexts have no waiting caller to notify. } } 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 0fb36c77761..4d2e06ba75a 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 @@ -18,6 +18,7 @@ package org.apache.rocketmq.proxy.service.channel; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.remoting.protocol.RemotingCommand; @@ -25,6 +26,7 @@ import org.junit.Test; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class InvocationChannelTest { @@ -86,6 +88,10 @@ public void testClearExpireContextShouldCompleteResponseFutureExceptionally() { assertFalse(channel.isWritable()); assertTrue(responseFuture.isCompletedExceptionally()); + CompletionException exception = assertThrows(CompletionException.class, responseFuture::join); + assertTrue(exception.getCause().getMessage().contains("after 0 seconds")); + assertTrue(exception.getCause().getMessage().contains("remoteAddress=127.0.0.1:8080")); + assertTrue(exception.getCause().getMessage().contains("localAddress=127.0.0.1:8081")); } finally { ConfigurationManager.getProxyConfig().setChannelExpiredInSeconds(originalExpiredInSeconds); }