opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-19 02:23:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7429] clean channel map when CLIENT_UNREGISTER in proxy

Browse Source
This commit is contained in:
lk
2023-10-09 16:07:56 +08:00
committed by GitHub
parent b9ffe0f957
commit 3808387e13
2 changed files with 88 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncer.java
+20 -11
View File
@@ -18,6 +18,7 @@
package org.apache.rocketmq.proxy.service.sysmessage;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
@@ -73,16 +74,8 @@ public class HeartbeatSyncer extends AbstractSystemMessageSyncer {
);
this.consumerManager.appendConsumerIdsChangeListener(new ConsumerIdsChangeListener() {
@Override
public void handle(ConsumerGroupEvent event, String s, Object... args) {
if (event == ConsumerGroupEvent.CLIENT_UNREGISTER) {
if (args == null || args.length < 1) {
return;
}
if (args[0] instanceof ClientChannelInfo) {
ClientChannelInfo clientChannelInfo = (ClientChannelInfo) args[0];
remoteChannelMap.remove(clientChannelInfo.getChannel().id().asLongText());
}
}
public void handle(ConsumerGroupEvent event, String group, Object... args) {
processConsumerGroupEvent(event, group, args);
}
@Override
@@ -98,6 +91,18 @@ public class HeartbeatSyncer extends AbstractSystemMessageSyncer {
super.shutdown();
}
protected void processConsumerGroupEvent(ConsumerGroupEvent event, String group, Object... args) {
if (event == ConsumerGroupEvent.CLIENT_UNREGISTER) {
if (args == null || args.length < 1) {
return;
}
if (args[0] instanceof ClientChannelInfo) {
ClientChannelInfo clientChannelInfo = (ClientChannelInfo) args[0];
remoteChannelMap.remove(buildKey(group, clientChannelInfo.getChannel()));
}
}
}
public void onConsumerRegister(String consumerGroup, ClientChannelInfo clientChannelInfo,
ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
Set<SubscriptionData> subList) {
@@ -189,7 +194,7 @@ public class HeartbeatSyncer extends AbstractSystemMessageSyncer {
}
RemoteChannel decodedChannel = RemoteChannel.decode(data.getChannelData());
RemoteChannel channel = remoteChannelMap.computeIfAbsent(data.getGroup() + "@" + decodedChannel.id().asLongText(), key -> decodedChannel);
RemoteChannel channel = remoteChannelMap.computeIfAbsent(buildKey(data.getGroup(), decodedChannel), key -> decodedChannel);
channel.setExtendAttribute(decodedChannel.getChannelExtendAttribute());
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
channel,
@@ -228,4 +233,8 @@ public class HeartbeatSyncer extends AbstractSystemMessageSyncer {
// use local address, remoting port and grpc port to build unique local proxy Id
return proxyConfig.getLocalServeAddr() + "%" + proxyConfig.getRemotingListenPort() + "%" + proxyConfig.getGrpcServerPort();
}
private static String buildKey(String group, Channel channel) {
return group + "@" + channel.id().asLongText();
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/service/sysmessage/HeartbeatSyncerTest.java
+68
View File
@@ -27,6 +27,7 @@ import com.google.common.collect.Sets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -35,6 +36,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupEvent;
import org.apache.rocketmq.broker.client.ConsumerManager;
import org.apache.rocketmq.client.impl.mqclient.MQClientAPIExt;
import org.apache.rocketmq.client.impl.mqclient.MQClientAPIFactory;
@@ -320,6 +322,72 @@ public class HeartbeatSyncerTest extends InitConfigTest {
}
}
@Test
public void testProcessConsumerGroupEventForRemoting() {
String consumerGroup = "consumerGroup";
Channel channel = createMockChannel();
RemotingProxyOutClient remotingProxyOutClient = mock(RemotingProxyOutClient.class);
RemotingChannel remotingChannel = new RemotingChannel(remotingProxyOutClient, proxyRelayService, channel, clientId, Collections.emptySet());
ClientChannelInfo clientChannelInfo = new ClientChannelInfo(
remotingChannel,
clientId,
LanguageCode.JAVA,
4
);
testProcessConsumerGroupEvent(consumerGroup, clientChannelInfo);
}
@Test
public void testProcessConsumerGroupEventForGrpcV2() {
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
);
testProcessConsumerGroupEvent(consumerGroup, clientChannelInfo);
}
private void testProcessConsumerGroupEvent(String consumerGroup, ClientChannelInfo clientChannelInfo) {
HeartbeatSyncer heartbeatSyncer = new HeartbeatSyncer(topicRouteService, adminService, consumerManager, mqClientAPIFactory, null);
SendResult okSendResult = new SendResult();
okSendResult.setSendStatus(SendStatus.SEND_OK);
ArgumentCaptor<Message> messageArgumentCaptor = ArgumentCaptor.forClass(Message.class);
doReturn(CompletableFuture.completedFuture(okSendResult)).when(this.mqClientAPIExt)
.sendMessageAsync(anyString(), anyString(), messageArgumentCaptor.capture(), any(), anyLong());
heartbeatSyncer.onConsumerRegister(
consumerGroup,
clientChannelInfo,
ConsumeType.CONSUME_PASSIVELY,
MessageModel.CLUSTERING,
ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
Collections.emptySet()
);
await().atMost(Duration.ofSeconds(3)).until(() -> messageArgumentCaptor.getAllValues().size() == 1);
// change local serve addr, to simulate other proxy receive messages
heartbeatSyncer.localProxyId = RandomStringUtils.randomAlphabetic(10);
ArgumentCaptor<ClientChannelInfo> channelInfoArgumentCaptor = ArgumentCaptor.forClass(ClientChannelInfo.class);
doReturn(true).when(consumerManager).registerConsumer(anyString(), channelInfoArgumentCaptor.capture(), any(), any(), any(), any(), anyBoolean());
heartbeatSyncer.consumeMessage(convertFromMessage(messageArgumentCaptor.getAllValues()), null);
assertEquals(1, heartbeatSyncer.remoteChannelMap.size());
heartbeatSyncer.processConsumerGroupEvent(ConsumerGroupEvent.CLIENT_UNREGISTER, consumerGroup, channelInfoArgumentCaptor.getValue());
assertTrue(heartbeatSyncer.remoteChannelMap.isEmpty());
}
private MessageExt convertFromMessage(Message message) {
MessageExt messageExt = new MessageExt();
messageExt.setTopic(message.getTopic());