diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java index 55cabe6f5e5..a51d3c6e9e3 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java +++ b/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 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); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java index 5d7b97f2296..e4c900e4fcc 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java +++ b/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)); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java index fa7e9982e1f..5d01d155441 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java +++ b/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 } diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java index 3ab78fba5f9..9cf176602e0 100644 --- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java +++ b/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; diff --git a/common/src/test/java/org/apache/rocketmq/common/KeyBuilderTest.java b/common/src/test/java/org/apache/rocketmq/common/KeyBuilderTest.java index 47191c907fe..41c38f75003 100644 --- a/common/src/test/java/org/apache/rocketmq/common/KeyBuilderTest.java +++ b/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(); + } } \ No newline at end of file diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java index b61c22b441e..0caeb9cf730 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ConsumerProcessorTest.java +++ b/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());