(rocketmq) branch develop updated: [ISSUE #11493] Avoid retaining channels for orphan unregister (#11494)

[fuyou]

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

fuyou001 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 592d59d12d [ISSUE #11493] Avoid retaining channels for orphan 
unregister (#11494)
592d59d12d is described below

commit 592d59d12db356f8c98332d9e355a89d92590db3
Author: qianye <wuxingcan....@alibaba-inc.com>
AuthorDate: Tue Oct 27 17:01:03 2026 +0800

    [ISSUE #11493] Avoid retaining channels for orphan unregister (#11494)
    
    * fix(proxy): avoid retaining channels for orphan unregister
    
    Only cache remote channels for REGISTER heartbeats so an unmatched
    UNREGISTER cannot leave an unreachable entry in remoteChannelMap.
    
    * fix(proxy): preserve unregister delivery for uncached channels
    
    Use the decoded channel without caching it when an UNREGISTER arrives
    for a channel absent from remoteChannelMap, so ConsumerManager can
    still perform its normal cleanup.
---
 .../proxy/service/sysmessage/HeartbeatSyncer.java  |  8 +++++-
 .../service/sysmessage/HeartbeatSyncerTest.java    | 32 ++++++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)

diff --git 
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java
 
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java
index 5d9fbb3069..9703c95fb9 100644
--- 
a/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java
+++ 
b/proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java
@@ -202,7 +202,13 @@ public class HeartbeatSyncer extends 
AbstractSystemMessageSyncer {
                 }
 
                 RemoteChannel decodedChannel = 
RemoteChannel.decode(data.getChannelData());
-                RemoteChannel channel = 
remoteChannelMap.computeIfAbsent(buildKey(data.getGroup(), decodedChannel), key 
-> decodedChannel);
+                String key = buildKey(data.getGroup(), decodedChannel);
+                RemoteChannel channel;
+                if (data.getHeartbeatType().equals(HeartbeatType.REGISTER)) {
+                    channel = remoteChannelMap.computeIfAbsent(key, k -> 
decodedChannel);
+                } else {
+                    channel = remoteChannelMap.getOrDefault(key, 
decodedChannel);
+                }
                 
channel.setExtendAttribute(decodedChannel.getChannelExtendAttribute());
                 ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
                     channel,
diff --git 
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java
 
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java
index 9a2c5e3437..d322502b64 100644
--- 
a/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java
+++ 
b/proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java
@@ -217,6 +217,38 @@ public class HeartbeatSyncerTest extends InitConfigTest {
         assertSame(channelInfoList.get(0).getChannel(), 
syncUnRegisterChannelInfoArgumentCaptor.getValue().getChannel());
     }
 
+    @Test
+    public void testSyncUnregisterWithoutRegister() throws Exception {
+        String consumerGroup = "consumerGroup";
+        GrpcClientSettingsManager grpcClientSettingsManager = 
mock(GrpcClientSettingsManager.class);
+        GrpcChannelManager grpcChannelManager = mock(GrpcChannelManager.class);
+        GrpcClientChannel grpcClientChannel = new GrpcClientChannel(
+            proxyRelayService, grpcClientSettingsManager, grpcChannelManager,
+            
ProxyContext.create().setRemoteAddress(remoteAddress).setLocalAddress(localAddress),
+            clientId);
+        ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
+            grpcClientChannel,
+            clientId,
+            LanguageCode.JAVA,
+            5
+        );
+        HeartbeatSyncer heartbeatSyncer = new 
HeartbeatSyncer(topicRouteService, adminService, consumerManager, 
mqClientAPIFactory, null);
+
+        ArgumentCaptor<Message> messageArgumentCaptor = 
ArgumentCaptor.forClass(Message.class);
+        SendResult sendResult = new SendResult();
+        sendResult.setSendStatus(SendStatus.SEND_OK);
+        
doReturn(CompletableFuture.completedFuture(sendResult)).when(this.mqClientAPIExt)
+            .sendMessageAsync(anyString(), anyString(), 
messageArgumentCaptor.capture(), any(), anyLong());
+        heartbeatSyncer.onConsumerUnRegister(consumerGroup, clientChannelInfo);
+        await().atMost(Duration.ofSeconds(3)).until(() -> 
!messageArgumentCaptor.getAllValues().isEmpty());
+
+        heartbeatSyncer.localProxyId = RandomStringUtils.randomAlphabetic(10);
+        
heartbeatSyncer.consumeMessage(Lists.newArrayList(convertFromMessage(messageArgumentCaptor.getValue())),
 null);
+
+        assertTrue(heartbeatSyncer.remoteChannelMap.isEmpty());
+        verify(consumerManager).unregisterConsumer(eq(consumerGroup), any(), 
eq(false));
+    }
+
     @Test
     public void testSyncRemotingChannel() throws Exception {
         String consumerGroup = "consumerGroup";