(rocketmq) branch develop updated: [ISSUE #10247] Remove duplicate remove call in InvocationChannel (#10248)

[yuzhou]

This is an automated email from the ASF dual-hosted git repository.

yuz10 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 079038dafc [ISSUE #10247] Remove duplicate remove call in 
InvocationChannel (#10248)
079038dafc is described below

commit 079038dafc34fc75a46b5eef51593fce5d012786
Author: yx9o <yangx_s...@163.com>
AuthorDate: Thu May 21 15:17:27 2026 +0800

    [ISSUE #10247] Remove duplicate remove call in InvocationChannel (#10248)
---
 .../proxy/service/channel/InvocationChannel.java   |  1 -
 .../service/channel/InvocationChannelTest.java     | 67 ++++++++++++++++++++++
 2 files changed, 67 insertions(+), 1 deletion(-)

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 00e8cea99c..bbaaddd293 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
@@ -41,7 +41,6 @@ public class InvocationChannel extends SimpleChannel {
             if (null != context) {
                 context.handle(responseCommand);
             }
-            inFlightRequestMap.remove(responseCommand.getOpaque());
         }
         return super.writeAndFlush(msg);
     }
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
new file mode 100644
index 0000000000..ddede4fbc8
--- /dev/null
+++ 
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/channel/InvocationChannelTest.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.service.channel;
+
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+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 {
+
+    @Test
+    public void testWriteAndFlushShouldNotRemoveReRegisteredContext() {
+        InvocationChannel channel = new InvocationChannel("127.0.0.1:8080", 
"127.0.0.1:8081");
+        AtomicBoolean nextContextHandled = new AtomicBoolean(false);
+
+        channel.registerInvocationContext(1, new InvocationContextInterface() {
+            @Override
+            public void handle(RemotingCommand remotingCommand) {
+                channel.registerInvocationContext(remotingCommand.getOpaque(), 
new InvocationContextInterface() {
+                    @Override
+                    public void handle(RemotingCommand nextRemotingCommand) {
+                        nextContextHandled.set(true);
+                    }
+
+                    @Override
+                    public boolean expired(long expiredTimeSec) {
+                        return false;
+                    }
+                });
+            }
+
+            @Override
+            public boolean expired(long expiredTimeSec) {
+                return false;
+            }
+        });
+
+        RemotingCommand response = RemotingCommand.createResponseCommand(0, 
"OK");
+        response.setOpaque(1);
+
+        channel.writeAndFlush(response);
+        assertTrue(channel.isWritable());
+
+        channel.writeAndFlush(response);
+        assertTrue(nextContextHandled.get());
+        assertFalse(channel.isWritable());
+    }
+}