(rocketmq) branch develop updated: [ISSUE #10661] Unregister all producer topics on gRPC client termination (#10662)

[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 31734b150b [ISSUE #10661] Unregister all producer topics on gRPC 
client termination (#10662)
31734b150b is described below

commit 31734b150b81460cfbb5b85df585e298046f52ca
Author: yx9o <yangx_s...@163.com>
AuthorDate: Mon Jul 27 17:33:06 2026 +0800

    [ISSUE #10661] Unregister all producer topics on gRPC client termination 
(#10662)
---
 .../rocketmq/proxy/grpc/v2/client/ClientActivity.java | 13 +++++++------
 .../proxy/grpc/v2/client/ClientActivityTest.java      | 19 +++++++++++++------
 2 files changed, 20 insertions(+), 12 deletions(-)

diff --git 
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
 
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
index b674d448c0..abc23a53a3 100644
--- 
a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
+++ 
b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java
@@ -156,16 +156,17 @@ public class ClientActivity extends 
AbstractMessagingActivity {
             }
 
             switch (clientSettings.getClientType()) {
-                case PRODUCER:
-                    for (Resource topic : 
clientSettings.getPublishing().getTopicsList()) {
-                        String topicName = topic.getName();
-                        GrpcClientChannel channel = 
this.grpcChannelManager.removeChannel(clientId);
-                        if (channel != null) {
-                            ClientChannelInfo clientChannelInfo = new 
ClientChannelInfo(channel, clientId, languageCode, 
MQVersion.Version.V5_0_0.ordinal());
+                case PRODUCER: {
+                    GrpcClientChannel channel = 
this.grpcChannelManager.removeChannel(clientId);
+                    if (channel != null) {
+                        ClientChannelInfo clientChannelInfo = new 
ClientChannelInfo(channel, clientId, languageCode, 
MQVersion.Version.V5_0_0.ordinal());
+                        for (Resource topic : 
clientSettings.getPublishing().getTopicsList()) {
+                            String topicName = topic.getName();
                             this.messagingProcessor.unRegisterProducer(ctx, 
topicName, clientChannelInfo);
                         }
                     }
                     break;
+                }
                 case PUSH_CONSUMER:
                 case LITE_PUSH_CONSUMER:
                 case SIMPLE_CONSUMER:
diff --git 
a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
 
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
index 532c9795c8..e215c6efab 100644
--- 
a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
+++ 
b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivityTest.java
@@ -224,18 +224,21 @@ public class ClientActivityTest extends BaseActivityTest {
     @Test
     public void testProducerNotifyClientTermination() throws Throwable {
         ProxyContext context = createContext();
+        String anotherTopic = "anotherTopic";
 
-        
when(this.grpcClientSettingsManager.removeAndGetClientSettings(any())).thenReturn(Settings.newBuilder()
+        Settings settings = Settings.newBuilder()
             .setClientType(ClientType.PRODUCER)
             .setPublishing(Publishing.newBuilder()
                 .addTopics(Resource.newBuilder().setName(TOPIC).build())
+                .addTopics(Resource.newBuilder().setName(anotherTopic).build())
                 .build())
-            .build());
+            .build();
+        
when(this.grpcClientSettingsManager.removeAndGetClientSettings(any())).thenReturn(settings);
+        ArgumentCaptor<String> topicArgumentCaptor = 
ArgumentCaptor.forClass(String.class);
         ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = 
ArgumentCaptor.forClass(ClientChannelInfo.class);
-        doNothing().when(this.messagingProcessor).unRegisterProducer(any(), 
anyString(), channelInfoArgumentCaptor.capture());
         when(this.metadataService.getTopicMessageType(any(), 
anyString())).thenReturn(TopicMessageType.NORMAL);
 
-        this.sendProducerTelemetry(context);
+        this.sendClientTelemetry(context, settings);
         this.sendProducerHeartbeat(context);
 
         NotifyClientTerminationResponse response = 
this.clientActivity.notifyClientTermination(
@@ -245,8 +248,12 @@ public class ClientActivityTest extends BaseActivityTest {
         ).get();
 
         assertEquals(Code.OK, response.getStatus().getCode());
-        ClientChannelInfo clientChannelInfo = 
channelInfoArgumentCaptor.getValue();
-        assertClientChannelInfo(clientChannelInfo, TOPIC);
+        verify(this.messagingProcessor, times(2)).unRegisterProducer(any(), 
topicArgumentCaptor.capture(),
+            channelInfoArgumentCaptor.capture());
+        assertThat(topicArgumentCaptor.getAllValues()).containsExactly(TOPIC, 
anotherTopic);
+        assertThat(channelInfoArgumentCaptor.getAllValues().get(0))
+            .isSameAs(channelInfoArgumentCaptor.getAllValues().get(1));
+        assertClientChannelInfo(channelInfoArgumentCaptor.getValue(), TOPIC);
     }
 
     @Test