From 9ca11e98eb4b0404a2d0e12052ba22a433b0a686 Mon Sep 17 00:00:00 2001 From: zhouxiang Date: Tue, 2 Jul 2024 20:02:54 +0800 Subject: [PATCH] [ISSUE #8352] Fix CLIENT_REGISTER in registerConsumer --- .../org/apache/rocketmq/broker/client/ConsumerManager.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java index 42e71e7e997..9f838b51544 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java @@ -178,8 +178,6 @@ public boolean registerConsumer(final String group, final ClientChannelInfo clie long start = System.currentTimeMillis(); ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group); if (null == consumerGroupInfo) { - callConsumerIdsChangeListener(ConsumerGroupEvent.CLIENT_REGISTER, group, clientChannelInfo, - subList.stream().map(SubscriptionData::getTopic).collect(Collectors.toSet())); ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere); ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp); consumerGroupInfo = prev != null ? prev : tmp; @@ -188,6 +186,10 @@ public boolean registerConsumer(final String group, final ClientChannelInfo clie boolean r1 = consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel, consumeFromWhere); + if (r1) { + callConsumerIdsChangeListener(ConsumerGroupEvent.CLIENT_REGISTER, group, clientChannelInfo, + subList.stream().map(SubscriptionData::getTopic).collect(Collectors.toSet())); + } boolean r2 = false; if (updateSubscription) { r2 = consumerGroupInfo.updateSubscription(subList);