Merge pull request #466 from zhangjidi2016/fix_broadcast

[ISSUE #458]Fixed the problem of cannot consume previous messages in broadcast consumption mode
apache · Jul 23, 2022 · c3b851f · c3b851f
2 parents 1aae0cd + a11049d
commit c3b851f
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 5 deletions.
diff --git a/...ot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java b/...ot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
@@ -619,7 +619,6 @@ private void initRocketMQPushConsumer() throws MQClientException {
                     resolveRequiredPlaceholders(this.rocketMQMessageListener.customizedTraceTopic()));
         }
         consumer.setNamespace(namespace);
-        consumer.setInstanceName(RocketMQUtil.getInstanceName(nameServer));
 
         String customizedNameServer = this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.rocketMQMessageListener.nameServer());
         if (customizedNameServer != null) {
@@ -639,9 +638,11 @@ private void initRocketMQPushConsumer() throws MQClientException {
         switch (messageModel) {
             case BROADCASTING:
                 consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
+                consumer.setInstanceName(Long.toString(nameServer.hashCode()));
                 break;
             case CLUSTERING:
                 consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
+                consumer.setInstanceName(RocketMQUtil.getInstanceName(nameServer));
                 break;
             default:
                 throw new IllegalArgumentException("Property 'messageModel' was wrong.");

diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQUtil.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQUtil.java
@@ -293,7 +293,7 @@ public static String getInstanceName(String identify) {
             instanceName.append(identify, 0, maxLength)
                     .append(identify.hashCode());
         } else {
-            instanceName.append(identify);
+            instanceName.append(identify.hashCode());
         }
         instanceName.append(separator).append(UtilAll.getPid())
                 .append(separator).append(System.nanoTime());
@@ -312,7 +312,6 @@ public static DefaultLitePullConsumer createDefaultLitePullConsumer(String nameS
             litePullConsumer = new DefaultLitePullConsumer(groupName);
         }
         litePullConsumer.setNamesrvAddr(nameServer);
-        litePullConsumer.setInstanceName(RocketMQUtil.getInstanceName(nameServer));
         litePullConsumer.setPullBatchSize(pullBatchSize);
         if (accessChannel != null) {
             litePullConsumer.setAccessChannel(AccessChannel.valueOf(accessChannel));
@@ -322,9 +321,11 @@ public static DefaultLitePullConsumer createDefaultLitePullConsumer(String nameS
         switch (messageModel) {
             case BROADCASTING:
                 litePullConsumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
+                litePullConsumer.setInstanceName(Long.toString(nameServer.hashCode()));
                 break;
             case CLUSTERING:
                 litePullConsumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
+                litePullConsumer.setInstanceName(RocketMQUtil.getInstanceName(nameServer));
                 break;
             default:
                 throw new IllegalArgumentException("Property 'messageModel' was wrong.");

diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/RocketMQUtilTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/RocketMQUtilTest.java
@@ -148,8 +148,8 @@ public void testConvertToSpringMessage() {
     @Test
     public void testGetInstanceName() {
         String nameServer = "127.0.0.1:9876";
-        String expected = "127.0.0.1:9876@";
-        assertEquals(expected + UtilAll.getPid(), removeNanoTime(RocketMQUtil.getInstanceName(nameServer)));
+        String expected = "127.0.0.1:9876";
+        assertEquals(expected.hashCode() + "@" + UtilAll.getPid(), removeNanoTime(RocketMQUtil.getInstanceName(nameServer)));
 
         nameServer = "I-am-a-very-very-long-domain-name-1:9876;I-am-a-very-very-long-domain-name-2:9876;I-am-a-very-very-long-domain-name-3:9876";
         expected = "I-am-a-very-very-long-domain-name-1:9876;I-am-a-very-very-long-domain-name-2:9876;I-am-a-very-very-l-335144505@";