apache · totalo · Dec 24, 2023 · Dec 24, 2023 · Feb 4, 2025 · Feb 7, 2025
diff --git a/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java b/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java
@@ -154,9 +154,9 @@ public void setInstanceName(String instanceName) {
         this.instanceName = instanceName;
     }
 
-    public void changeInstanceNameToPID() {
+    public void changeInstanceNameToIpWithPidAndGroupInfo(String groupInfo) {
         if (this.instanceName.equals("DEFAULT")) {
-            this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
+            this.instanceName = this.clientIP + "#" + UtilAll.getPid() + "#" + groupInfo;
         }
     }
 

diff --git a/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
@@ -286,7 +286,7 @@ public synchronized void start() throws MQClientException {
                 this.checkConfig();
 
                 if (this.defaultLitePullConsumer.getMessageModel() == MessageModel.CLUSTERING) {
-                    this.defaultLitePullConsumer.changeInstanceNameToPID();
+                    this.defaultLitePullConsumer.changeInstanceNameToIpWithPidAndGroupInfo(this.defaultLitePullConsumer.getConsumerGroup());
                 }
 
                 initScheduledThreadPoolExecutor();

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
@@ -710,7 +710,7 @@ public synchronized void start() throws MQClientException {
                 this.copySubscription();
 
                 if (this.defaultMQPullConsumer.getMessageModel() == MessageModel.CLUSTERING) {
-                    this.defaultMQPullConsumer.changeInstanceNameToPID();
+                    this.defaultMQPullConsumer.changeInstanceNameToIpWithPidAndGroupInfo(this.defaultMQPullConsumer.getConsumerGroup());
                 }
 
                 this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPullConsumer, this.rpcHook);

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
@@ -932,7 +932,7 @@ public synchronized void start() throws MQClientException {
                 this.copySubscription();
 
                 if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
-                    this.defaultMQPushConsumer.changeInstanceNameToPID();
+                    this.defaultMQPushConsumer.changeInstanceNameToIpWithPidAndGroupInfo(this.defaultMQPushConsumer.getConsumerGroup());
                 }
 
                 this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@ -248,9 +248,8 @@ public void start(final boolean startFactory) throws MQClientException {
                 this.checkConfig();
 
                 if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
-                    this.defaultMQProducer.changeInstanceNameToPID();
+                    this.defaultMQProducer.changeInstanceNameToIpWithPidAndGroupInfo(this.defaultMQProducer.getProducerGroup());
                 }
-
                 this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
 
                 defaultMQProducer.initProduceAccumulator();

diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
@@ -892,7 +892,7 @@ private PullResultExt createPullResult(PullMessageRequestHeader requestHeader, P
     private void suppressUpdateTopicRouteInfoFromNameServer(
         DefaultLitePullConsumer litePullConsumer) throws IllegalAccessException {
         if (litePullConsumer.getMessageModel() == MessageModel.CLUSTERING) {
-            litePullConsumer.changeInstanceNameToPID();
+            litePullConsumer.changeInstanceNameToIpWithPidAndGroupInfo(litePullConsumer.getConsumerGroup());
         }
 
         ConcurrentMap<String, MQClientInstance> factoryTable = (ConcurrentMap<String, MQClientInstance>) FieldUtils.readDeclaredField(MQClientManager.getInstance(), "factoryTable", true);

diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
@@ -152,7 +152,7 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
         DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
 
         // suppress updateTopicRouteInfoFromNameServer
-        pushConsumer.changeInstanceNameToPID();
+        pushConsumer.changeInstanceNameToIpWithPidAndGroupInfo(pushConsumer.getConsumerGroup());
         mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(pushConsumer, (RPCHook) FieldUtils.readDeclaredField(pushConsumerImpl, "rpcHook", true));
         FieldUtils.writeDeclaredField(mQClientFactory, "mQClientAPIImpl", mQClientAPIImpl, true);
         mQClientFactory = spy(mQClientFactory);

diff --git a/.../java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyServiceTest.java b/.../java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyServiceTest.java
@@ -118,7 +118,7 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeCo
         field.set(pushConsumerImpl, rebalancePushImpl);
         pushConsumer.subscribe(topic, "*");
         // suppress updateTopicRouteInfoFromNameServer
-        pushConsumer.changeInstanceNameToPID();
+        pushConsumer.changeInstanceNameToIpWithPidAndGroupInfo(pushConsumer.getConsumerGroup());
         mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(pushConsumer, (RPCHook) FieldUtils.readDeclaredField(pushConsumerImpl, "rpcHook", true));
         mQClientFactory = spy(mQClientFactory);
         field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");

diff --git a/.../test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyServiceTest.java b/.../test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyServiceTest.java
@@ -109,7 +109,7 @@ public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderly
         pushConsumer.subscribe(topic, "*");
 
         // suppress updateTopicRouteInfoFromNameServer
-        pushConsumer.changeInstanceNameToPID();
+        pushConsumer.changeInstanceNameToIpWithPidAndGroupInfo(pushConsumer.getConsumerGroup());
         mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(pushConsumer, (RPCHook) FieldUtils.readDeclaredField(pushConsumerImpl, "rpcHook", true));
         mQClientFactory = spy(mQClientFactory);
         field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");

diff --git a/.../src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithOpenTracingTest.java b/.../src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithOpenTracingTest.java
@@ -153,7 +153,7 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
         DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
 
         // suppress updateTopicRouteInfoFromNameServer
-        pushConsumer.changeInstanceNameToPID();
+        pushConsumer.changeInstanceNameToIpWithPidAndGroupInfo(pushConsumer.getConsumerGroup());
         mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(pushConsumer, (RPCHook) FieldUtils.readDeclaredField(pushConsumerImpl, "rpcHook", true));
         FieldUtils.writeDeclaredField(mQClientFactory, "mQClientAPIImpl", mQClientAPIImpl, true);
         mQClientFactory = spy(mQClientFactory);

diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java
@@ -142,7 +142,7 @@ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
         DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
 
         // suppress updateTopicRouteInfoFromNameServer
-        pushConsumer.changeInstanceNameToPID();
+        pushConsumer.changeInstanceNameToIpWithPidAndGroupInfo(pushConsumer.getConsumerGroup());
         mQClientFactory = spy(MQClientManager.getInstance().getOrCreateMQClientInstance(pushConsumer, (RPCHook) FieldUtils.readDeclaredField(pushConsumerImpl, "rpcHook", true)));
         factoryTable.put(pushConsumer.buildMQClientId(), mQClientFactory);
         doReturn(false).when(mQClientFactory).updateTopicRouteInfoFromNameServer(anyString());

diff --git a/...rc/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java b/...rc/test/java/org/apache/rocketmq/client/trace/DefaultMQLitePullConsumerWithTraceTest.java
@@ -308,7 +308,7 @@ private SendResult createSendResult(SendStatus sendStatus) {
     private static void suppressUpdateTopicRouteInfoFromNameServer(DefaultLitePullConsumer litePullConsumer) throws IllegalAccessException {
         DefaultLitePullConsumerImpl defaultLitePullConsumerImpl = (DefaultLitePullConsumerImpl) FieldUtils.readDeclaredField(litePullConsumer, "defaultLitePullConsumerImpl", true);
         if (litePullConsumer.getMessageModel() == MessageModel.CLUSTERING) {
-            litePullConsumer.changeInstanceNameToPID();
+            litePullConsumer.changeInstanceNameToIpWithPidAndGroupInfo(litePullConsumer.getConsumerGroup());
         }
         MQClientInstance mQClientFactory = spy(MQClientManager.getInstance().getOrCreateMQClientInstance(litePullConsumer, (RPCHook) FieldUtils.readDeclaredField(defaultLitePullConsumerImpl, "rpcHook", true)));
         ConcurrentMap<String, MQClientInstance> factoryTable = (ConcurrentMap<String, MQClientInstance>) FieldUtils.readDeclaredField(MQClientManager.getInstance(), "factoryTable", true);

diff --git a/example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java b/example/src/main/java/org/apache/rocketmq/example/quickstart/Consumer.java
@@ -27,7 +27,7 @@
  */
 public class Consumer {
 
-    public static final String CONSUMER_GROUP = "please_rename_unique_group_name_4";
+    public static final String CONSUMER_GROUP = "c_please_rename_unique_group_name_4";
     public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
     public static final String TOPIC = "TopicTest";
 
@@ -37,6 +37,7 @@ public static void main(String[] args) throws MQClientException {
          * Instantiate with specified consumer group name.
          */
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
+        DefaultMQPushConsumer consumer2 = new DefaultMQPushConsumer(CONSUMER_GROUP);
 
         /*
          * Specify name server addresses.
@@ -50,17 +51,21 @@ public static void main(String[] args) throws MQClientException {
          * </pre>
          */
         // Uncomment the following line while debugging, namesrvAddr should be set to your local address
-        // consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
+        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
+        consumer2.setNamesrvAddr(DEFAULT_NAMESRVADDR);
 
         /*
          * Specify where to start in case the specific consumer group is a brand-new one.
          */
         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+        consumer2.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 
         /*
          * Subscribe one more topic to consume.
          */
         consumer.subscribe(TOPIC, "*");
+        consumer.subscribe(TOPIC + "1", "*");
+        consumer2.subscribe(TOPIC, "*");
 
         /*
          *  Register callback to execute on arrival of messages fetched from brokers.
@@ -69,11 +74,16 @@ public static void main(String[] args) throws MQClientException {
             System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
         });
+        consumer2.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
+            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
 
         /*
          *  Launch the consumer instance.
          */
         consumer.start();
+        consumer2.start();
 
         System.out.printf("Consumer Started.%n");
     }

diff --git a/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java b/example/src/main/java/org/apache/rocketmq/example/quickstart/Producer.java
@@ -31,7 +31,7 @@ public class Producer {
      * The number of produced messages.
      */
     public static final int MESSAGE_COUNT = 1000;
-    public static final String PRODUCER_GROUP = "please_rename_unique_group_name";
+    public static final String PRODUCER_GROUP = "p_please_rename_unique_group_name";
     public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
     public static final String TOPIC = "TopicTest";
     public static final String TAG = "TagA";

diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/CombineConsumeQueueStore.java b/store/src/main/java/org/apache/rocketmq/store/queue/CombineConsumeQueueStore.java
@@ -31,15 +31,15 @@
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.message.MessageExtBrokerInner;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.store.DefaultMessageStore;
 import org.apache.rocketmq.store.DispatchRequest;
 import org.apache.rocketmq.store.StoreType;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.exception.ConsumeQueueException;
 import org.apache.rocketmq.store.exception.StoreException;
 import org.rocksdb.RocksDBException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class CombineConsumeQueueStore implements ConsumeQueueStoreInterface {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

diff --git a/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java b/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
@@ -64,6 +64,7 @@ public class MQAdminTestUtils {
     public static void startAdmin(String nameSrvAddr) throws MQClientException {
         mqAdminExt = new DefaultMQAdminExt();
         mqAdminExt.setNamesrvAddr(nameSrvAddr);
+        mqAdminExt.setInstanceName(UUID.randomUUID().toString());
         mqAdminExt.start();
     }
 

diff --git a/test/src/test/resources/schema/api/client.consumer.DefaultLitePullConsumer.schema b/test/src/test/resources/schema/api/client.consumer.DefaultLitePullConsumer.schema
@@ -63,7 +63,7 @@ Field useTLS : private boolean false
 Field vipChannelEnabled : private boolean false
 Method assign(java.util.Collection) : public throws (void)
 Method buildMQClientId() : public throws (java.lang.String)
-Method changeInstanceNameToPID() : public throws (void)
+Method changeInstanceNameToIpWithPidAndGroupInfo(java.lang.String) : public throws (void)
 Method cloneClientConfig() : public throws (org.apache.rocketmq.client.ClientConfig)
 Method commit(boolean,java.util.Set) : public throws (void)
 Method commitSync() : public throws (void)

diff --git a/test/src/test/resources/schema/api/client.consumer.DefaultMQPullConsumer.schema b/test/src/test/resources/schema/api/client.consumer.DefaultMQPullConsumer.schema
@@ -47,7 +47,7 @@ Field unitName : private java.lang.String null
 Field useTLS : private boolean false
 Field vipChannelEnabled : private boolean false
 Method buildMQClientId() : public throws (java.lang.String)
-Method changeInstanceNameToPID() : public throws (void)
+Method changeInstanceNameToIpWithPidAndGroupInfo(java.lang.String) : public throws (void)
 Method cloneClientConfig() : public throws (org.apache.rocketmq.client.ClientConfig)
 Method createTopic(int,int,java.lang.String,java.lang.String) : public throws (void)
 Method createTopic(int,java.lang.String,java.lang.String) : public throws (void)

diff --git a/test/src/test/resources/schema/api/client.consumer.DefaultMQPushConsumer.schema b/test/src/test/resources/schema/api/client.consumer.DefaultMQPushConsumer.schema
@@ -63,7 +63,7 @@ Field unitName : private java.lang.String null
 Field useTLS : private boolean false
 Field vipChannelEnabled : private boolean false
 Method buildMQClientId() : public throws (java.lang.String)
-Method changeInstanceNameToPID() : public throws (void)
+Method changeInstanceNameToIpWithPidAndGroupInfo(java.lang.String) : public throws (void)
 Method cloneClientConfig() : public throws (org.apache.rocketmq.client.ClientConfig)
 Method createTopic(int,int,java.lang.String,java.lang.String) : public throws (void)
 Method createTopic(int,java.lang.String,java.lang.String) : public throws (void)

diff --git a/test/src/test/resources/schema/api/client.producer.DefaultMQProducer.schema b/test/src/test/resources/schema/api/client.producer.DefaultMQProducer.schema
@@ -50,7 +50,7 @@ Field useTLS : private boolean false
 Field vipChannelEnabled : private boolean false
 Method addRetryResponseCode(int) : public throws (void)
 Method buildMQClientId() : public throws (java.lang.String)
-Method changeInstanceNameToPID() : public throws (void)
+Method changeInstanceNameToIpWithPidAndGroupInfo(java.lang.String) : public throws (void)
 Method cloneClientConfig() : public throws (org.apache.rocketmq.client.ClientConfig)
 Method createTopic(int,int,java.lang.String,java.lang.String) : public throws (void)
 Method createTopic(int,java.lang.String,java.lang.String) : public throws (void)

diff --git a/test/src/test/resources/schema/api/tools.admin.DefaultMQAdminExt.schema b/test/src/test/resources/schema/api/tools.admin.DefaultMQAdminExt.schema
@@ -41,7 +41,7 @@ Field useTLS : private boolean false
 Field vipChannelEnabled : private boolean false
 Method addWritePermOfBroker(java.lang.String,java.lang.String) : public throws (int)
 Method buildMQClientId() : public throws (java.lang.String)
-Method changeInstanceNameToPID() : public throws (void)
+Method changeInstanceNameToIpWithPidAndGroupInfo(java.lang.String) : public throws (void)
 Method cleanExpiredConsumerQueue(java.lang.String) : public throws (boolean)
 Method cleanExpiredConsumerQueueByAddr(java.lang.String) : public throws (boolean)
 Method cleanUnusedTopic(java.lang.String) : public throws (boolean)

diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -155,7 +155,7 @@ public void start() throws MQClientException {
             case CREATE_JUST:
                 this.serviceState = ServiceState.START_FAILED;
 
-                this.defaultMQAdminExt.changeInstanceNameToPID();
+                this.defaultMQAdminExt.changeInstanceNameToIpWithPidAndGroupInfo(this.defaultMQAdminExt.getAdminExtGroup());
 
                 if ("{}".equals(this.defaultMQAdminExt.getSocksProxyConfig())) {
                     String proxyConfig = System.getenv(SOCKS_PROXY_JSON);