[ISSUE #9633] Enable retry topic V2 by default to prevent name collision

broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java

@@ -520,11 +520,17 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingC
         long popTime = System.currentTimeMillis();
         CompletableFuture<Long> getMessageFuture = CompletableFuture.completedFuture(0L);
         if (needRetry && !requestHeader.isOrder()) {
-            if (needRetryV1) {
+            // In priority mode this is the only chance to serve retry messages ahead of the
+            // normal topic, so cover both retry topic naming schemes instead of alternating
+            // between them while the V1 compatibility window is open.
+            boolean coverBothRetryTopics = usePriorityMode && brokerConfig.isEnableRetryTopicV2()
+                && brokerConfig.isRetrieveMessageFromPopRetryTopicV1();
+            if (needRetryV1 || coverBothRetryTopics) {
                 String retryTopic = KeyBuilder.buildPopRetryTopicV1(requestHeader.getTopic(), requestHeader.getConsumerGroup());
                 getMessageFuture = popMsgFromTopic(retryTopic, true, getMessageResult, requestHeader, reviveQid, channel,
                     popTime, finalMessageFilter, startOffsetInfo, msgOffsetInfo, orderCountInfo, randomQ, getMessageFuture);
-            } else {
+            }
+            if (!needRetryV1 || coverBothRetryTopics) {
                 String retryTopic = KeyBuilder.buildPopRetryTopic(requestHeader.getTopic(), requestHeader.getConsumerGroup(), brokerConfig.isEnableRetryTopicV2());
                 getMessageFuture = popMsgFromTopic(retryTopic, true, getMessageResult, requestHeader, reviveQid, channel,
                     popTime, finalMessageFilter, startOffsetInfo, msgOffsetInfo, orderCountInfo, randomQ, getMessageFuture);

broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java

@@ -166,8 +166,11 @@ public void testProcessRequest_whenTimerWheelIsFalse() throws RemotingCommandExc
     @Test
     public void testGetInitOffset_retryTopic() throws RemotingCommandException {
         when(messageStore.getMessageStoreConfig()).thenReturn(new MessageStoreConfig());
+        // pin the retry topic version so the pop path does not randomly switch between V1 and V2
+        brokerController.getBrokerConfig().setRetrieveMessageFromPopRetryTopicV1(false);
         String newGroup = group + "-" + System.currentTimeMillis();
-        String retryTopic = KeyBuilder.buildPopRetryTopic(topic, newGroup);
+        String retryTopic = KeyBuilder.buildPopRetryTopic(topic, newGroup,
+            brokerController.getBrokerConfig().isEnableRetryTopicV2());
         long minOffset = 100L;
         when(messageStore.getMinOffsetInQueue(retryTopic, 0)).thenReturn(minOffset);
         brokerController.getTopicConfigManager().getTopicConfigTable().put(retryTopic, new TopicConfig(retryTopic, 1, 1));

broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java

@@ -258,7 +258,7 @@ public void testReviveMsgFromCk_messageFound_writeRetryOK() throws Throwable {
         });
 
         popReviveService.mergeAndRevive(reviveObj);
-        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, false), actualRetryTopic.toString());
+        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, brokerConfig.isEnableRetryTopicV2()), actualRetryTopic.toString());
         verify(escapeBridge, times(1)).putMessageToSpecificQueue(any(MessageExtBrokerInner.class)); // write retry
         verify(messageStore, times(0)).putMessage(any(MessageExtBrokerInner.class)); // rewrite CK
     }
@@ -293,7 +293,7 @@ public void testReviveMsgFromCk_messageFound_writeRetryFailed_rewriteCK() throws
         // Wait for async operations to complete
         Thread.sleep(1000);
 
-        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, false), actualRetryTopic.toString());
+        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, brokerConfig.isEnableRetryTopicV2()), actualRetryTopic.toString());
         Assert.assertEquals(REVIVE_TOPIC, actualReviveTopic.toString());
         Assert.assertEquals(INVISIBLE_TIME + 10 * 1000L, actualInvisibleTime.get()); // first interval is 10s
         verify(escapeBridge, times(1)).putMessageToSpecificQueue(any(MessageExtBrokerInner.class)); // write retry
@@ -319,7 +319,7 @@ public void testReviveMsgFromCk_messageFound_writeRetryFailed_rewriteCK_end() th
         });
 
         popReviveService.mergeAndRevive(reviveObj);
-        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, false), actualRetryTopic.toString());
+        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, brokerConfig.isEnableRetryTopicV2()), actualRetryTopic.toString());
         verify(escapeBridge, times(1)).putMessageToSpecificQueue(any(MessageExtBrokerInner.class)); // write retry
         verify(messageStore, times(0)).putMessage(any(MessageExtBrokerInner.class)); // rewrite CK
     }
@@ -343,7 +343,7 @@ public void testReviveMsgFromCk_messageFound_writeRetryFailed_rewriteCK_noEnd()
         });
 
         popReviveService.mergeAndRevive(reviveObj);
-        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, false), actualRetryTopic.toString());
+        Assert.assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, GROUP, brokerConfig.isEnableRetryTopicV2()), actualRetryTopic.toString());
         verify(escapeBridge, times(1)).putMessageToSpecificQueue(any(MessageExtBrokerInner.class)); // write retry
         verify(messageStore, times(1)).putMessage(any(MessageExtBrokerInner.class)); // rewrite CK
     }

common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java

@@ -237,7 +237,7 @@ public class BrokerConfig extends BrokerIdentity {
     private boolean initPopOffsetByCheckMsgInMem = true;
     // read message from pop retry topic v1, for the compatibility, will be removed in the future version
     private boolean retrieveMessageFromPopRetryTopicV1 = true;
-    private boolean enableRetryTopicV2 = false;
+    private boolean enableRetryTopicV2 = true;
     private int popFromRetryProbability = 20;
     // pop retry probability for priority mode
     private int popFromRetryProbabilityForPriority = 0;

common/src/test/java/org/apache/rocketmq/common/KeyBuilderTest.java

@@ -60,4 +60,27 @@ public void testIsPopRetryTopicV2() {
         String popRetryTopicV1 = KeyBuilder.buildPopRetryTopicV1(topic, group);
         assertThat(KeyBuilder.isPopRetryTopicV2(popRetryTopicV1)).isEqualTo(false);
     }
+
+    @Test
+    public void testV1CollisionExample() {
+        // Demonstrates the V1 naming collision: different (group, topic) pairs produce the same retry topic
+        // group="A_B" topic="C" vs group="A" topic="B_C" both produce %RETRY%A_B_C
+        String collision1 = KeyBuilder.buildPopRetryTopicV1("C", "A_B");
+        String collision2 = KeyBuilder.buildPopRetryTopicV1("B_C", "A");
+        assertThat(collision1).isEqualTo(collision2); // both are %RETRY%A_B_C
+    }
+
+    @Test
+    public void testV2NoCollision() {
+        // V2 uses '+' separator which is not allowed in topic/group names, so no collision
+        String v2a = KeyBuilder.buildPopRetryTopicV2("C", "A_B");
+        String v2b = KeyBuilder.buildPopRetryTopicV2("B_C", "A");
+        assertThat(v2a).isNotEqualTo(v2b); // %RETRY%A_B+C vs %RETRY%A+B_C
+    }
+
+    @Test
+    public void testDefaultEnableRetryTopicV2IsTrue() {
+        BrokerConfig config = new BrokerConfig();
+        assertThat(config.isEnableRetryTopicV2()).isTrue();
+    }
 }

proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java

@@ -33,7 +33,6 @@
 import org.apache.rocketmq.client.consumer.PopResult;
 import org.apache.rocketmq.client.consumer.PopStatus;
 import org.apache.rocketmq.client.exception.MQBrokerException;
-import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.KeyBuilder;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.constant.ConsumeInitMode;
@@ -173,7 +172,8 @@ public void testAckMessage() throws Throwable {
             CONSUMER_GROUP, TOPIC, null, 3000).get();
 
         assertEquals(AckStatus.OK, ackResult.getStatus());
-        assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, CONSUMER_GROUP, new BrokerConfig().isEnableRetryTopicV2()), requestHeaderArgumentCaptor.getValue().getTopic());
+        // the retry topic version is encoded in the receipt handle (v1-shaped topic above), not taken from broker config
+        assertEquals(KeyBuilder.buildPopRetryTopicV1(TOPIC, CONSUMER_GROUP), requestHeaderArgumentCaptor.getValue().getTopic());
         assertEquals(CONSUMER_GROUP, requestHeaderArgumentCaptor.getValue().getConsumerGroup());
         assertEquals(handle.getReceiptHandle(), requestHeaderArgumentCaptor.getValue().getExtraInfo());
     }
@@ -296,7 +296,8 @@ public void testChangeInvisibleTime() throws Throwable {
             CONSUMER_GROUP, TOPIC, 1000, null, 3000, true).get();
 
         assertEquals(AckStatus.OK, ackResult.getStatus());
-        assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, CONSUMER_GROUP, new BrokerConfig().isEnableRetryTopicV2()), requestHeaderArgumentCaptor.getValue().getTopic());
+        // the retry topic version is encoded in the receipt handle (v1-shaped topic above), not taken from broker config
+        assertEquals(KeyBuilder.buildPopRetryTopicV1(TOPIC, CONSUMER_GROUP), requestHeaderArgumentCaptor.getValue().getTopic());
         assertEquals(CONSUMER_GROUP, requestHeaderArgumentCaptor.getValue().getConsumerGroup());
         assertEquals(1000, requestHeaderArgumentCaptor.getValue().getInvisibleTime().longValue());
         assertEquals(handle.getReceiptHandle(), requestHeaderArgumentCaptor.getValue().getExtraInfo());
@@ -456,7 +457,8 @@ public void testChangeInvisibleTimeWithSuspendFalse() throws Throwable {
             CONSUMER_GROUP, TOPIC, 1000, null, 3000, false).get();
 
         assertEquals(AckStatus.OK, ackResult.getStatus());
-        assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, CONSUMER_GROUP, new BrokerConfig().isEnableRetryTopicV2()), requestHeaderArgumentCaptor.getValue().getTopic());
+        // the retry topic version is encoded in the receipt handle (v1-shaped topic above), not taken from broker config
+        assertEquals(KeyBuilder.buildPopRetryTopicV1(TOPIC, CONSUMER_GROUP), requestHeaderArgumentCaptor.getValue().getTopic());
         assertEquals(CONSUMER_GROUP, requestHeaderArgumentCaptor.getValue().getConsumerGroup());
         assertEquals(1000, requestHeaderArgumentCaptor.getValue().getInvisibleTime().longValue());
         assertEquals(handle.getReceiptHandle(), requestHeaderArgumentCaptor.getValue().getExtraInfo());
@@ -478,7 +480,8 @@ public void testChangeInvisibleTimeWithSuspendTrue() throws Throwable {
             CONSUMER_GROUP, TOPIC, 1000, null, 3000, true).get();
 
         assertEquals(AckStatus.OK, ackResult.getStatus());
-        assertEquals(KeyBuilder.buildPopRetryTopic(TOPIC, CONSUMER_GROUP, new BrokerConfig().isEnableRetryTopicV2()), requestHeaderArgumentCaptor.getValue().getTopic());
+        // the retry topic version is encoded in the receipt handle (v1-shaped topic above), not taken from broker config
+        assertEquals(KeyBuilder.buildPopRetryTopicV1(TOPIC, CONSUMER_GROUP), requestHeaderArgumentCaptor.getValue().getTopic());
         assertEquals(CONSUMER_GROUP, requestHeaderArgumentCaptor.getValue().getConsumerGroup());
         assertEquals(1000, requestHeaderArgumentCaptor.getValue().getInvisibleTime().longValue());
         assertEquals(handle.getReceiptHandle(), requestHeaderArgumentCaptor.getValue().getExtraInfo());