opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #6686] Remove ClientManageProcessor#heartBeat repeat judgment (#6687)

Browse Source
This commit is contained in:
mxsm
2023-05-04 09:33:30 +08:00
committed by GitHub
parent 7903c70412
commit b4e6acca32
1 changed files with 28 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/ClientManageProcessor.java
+28 -29
View File
@@ -99,38 +99,37 @@ public class ClientManageProcessor implements NettyRequestProcessor {
}
}
SubscriptionGroupConfig subscriptionGroupConfig =
this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(
consumerData.getGroupName());
SubscriptionGroupConfig subscriptionGroupConfig = this.brokerController.getSubscriptionGroupManager()
.findSubscriptionGroupConfig(consumerData.getGroupName());
boolean isNotifyConsumerIdsChangedEnable = true;
if (null != subscriptionGroupConfig) {
isNotifyConsumerIdsChangedEnable = subscriptionGroupConfig.isNotifyConsumerIdsChangedEnable();
int topicSysFlag = 0;
if (consumerData.isUnitMode()) {
topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
}
String newTopic = MixAll.getRetryTopic(consumerData.getGroupName());
this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(
newTopic,
subscriptionGroupConfig.getRetryQueueNums(),
PermName.PERM_WRITE | PermName.PERM_READ, hasOrderTopicSub, topicSysFlag);
if (null == subscriptionGroupConfig) {
continue;
}
if (null != subscriptionGroupConfig) {
boolean changed = this.brokerController.getConsumerManager().registerConsumer(
consumerData.getGroupName(),
clientChannelInfo,
consumerData.getConsumeType(),
consumerData.getMessageModel(),
consumerData.getConsumeFromWhere(),
consumerData.getSubscriptionDataSet(),
isNotifyConsumerIdsChangedEnable
);
if (changed) {
LOGGER.info(
"ClientManageProcessor: registerConsumer info changed, SDK address={}, consumerData={}",
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), consumerData.toString());
}
isNotifyConsumerIdsChangedEnable = subscriptionGroupConfig.isNotifyConsumerIdsChangedEnable();
int topicSysFlag = 0;
if (consumerData.isUnitMode()) {
topicSysFlag = TopicSysFlag.buildSysFlag(false, true);
}
String newTopic = MixAll.getRetryTopic(consumerData.getGroupName());
this.brokerController.getTopicConfigManager().createTopicInSendMessageBackMethod(newTopic, subscriptionGroupConfig.getRetryQueueNums(),
PermName.PERM_WRITE | PermName.PERM_READ, hasOrderTopicSub, topicSysFlag);
boolean changed = this.brokerController.getConsumerManager().registerConsumer(
consumerData.getGroupName(),
clientChannelInfo,
consumerData.getConsumeType(),
consumerData.getMessageModel(),
consumerData.getConsumeFromWhere(),
consumerData.getSubscriptionDataSet(),
isNotifyConsumerIdsChangedEnable
);
if (changed) {
LOGGER.info("ClientManageProcessor: registerConsumer info changed, SDK address={}, consumerData={}",
RemotingHelper.parseChannelRemoteAddr(ctx.channel()), consumerData.toString());
}
}
for (ProducerData data : heartbeatData.getProducerDataSet()) {