[ISSUE #9443] Fix add wrong value to topicGroupTable in ConsumerManager (#9444)

Author: yx9o

broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerManager.java

@@ -292,7 +292,7 @@ public boolean registerConsumerWithoutSub(final String group, final ClientChanne
                 Set<String> prev = this.topicGroupTable.putIfAbsent(subscriptionData.getTopic(), tmp);
                 groups = prev != null ? prev : tmp;
             }
-            groups.add(subscriptionData.getTopic());
+            groups.add(group);
         }
 
         boolean updateChannelRst = consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel, consumeFromWhere);

broker/src/test/java/org/apache/rocketmq/broker/client/ConsumerManagerTest.java

@@ -37,6 +37,7 @@
 import java.util.HashSet;
 import java.util.Set;
 
+import static org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType.CONSUME_PASSIVELY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -115,7 +116,7 @@ public void registerConsumerTest() {
         final Set<SubscriptionData> subList = new HashSet<>();
         SubscriptionData subscriptionData = new SubscriptionData(TOPIC, "*");
         subList.add(subscriptionData);
-        consumerManager.registerConsumer(GROUP, clientChannelInfo, ConsumeType.CONSUME_PASSIVELY,
+        consumerManager.registerConsumer(GROUP, clientChannelInfo, CONSUME_PASSIVELY,
             MessageModel.BROADCASTING, ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET, subList, true);
         verify(consumerManager, never()).callConsumerIdsChangeListener(eq(ConsumerGroupEvent.CHANGE), any(), any());
         assertThat(consumerManager.getConsumerTable().get(GROUP)).isNotNull();
@@ -195,7 +196,7 @@ private void register() {
         final Set<SubscriptionData> subList = new HashSet<>();
         SubscriptionData subscriptionData = new SubscriptionData(TOPIC, "*");
         subList.add(subscriptionData);
-        consumerManager.registerConsumer(GROUP, clientChannelInfo, ConsumeType.CONSUME_PASSIVELY,
+        consumerManager.registerConsumer(GROUP, clientChannelInfo, CONSUME_PASSIVELY,
             MessageModel.BROADCASTING, ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET, subList, true);
     }
 
@@ -210,4 +211,24 @@ public void removeExpireConsumerGroupInfo() {
         assertThat(consumerManager.findSubscriptionData(GROUP, TOPIC)).isNull();
         assertThat(consumerManager.findSubscriptionData(GROUP, TOPIC + "_1")).isNotNull();
     }
+    
+    @Test
+    public void testRegisterConsumerWithoutSub() {
+        ConsumerGroupInfo groupInfo = new ConsumerGroupInfo(GROUP, CONSUME_PASSIVELY,
+                MessageModel.CLUSTERING, ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+        SubscriptionData subscriptionData = new SubscriptionData(TOPIC, "*");
+        groupInfo.getSubscriptionTable().put(TOPIC, subscriptionData);
+        consumerManager.getConsumerTable().put(GROUP, groupInfo);
+        
+        consumerManager.registerConsumerWithoutSub(GROUP,
+                clientChannelInfo,
+                CONSUME_PASSIVELY,
+                MessageModel.CLUSTERING,
+                ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET,
+                true);
+        
+        Set<String> actual = consumerManager.queryTopicConsumeByWho(TOPIC);
+        assertThat(actual).contains(GROUP);
+        assertThat(actual).doesNotContain(TOPIC);
+    }
 }