[ISSUE #9945] Use UniqueKey as TimerDelKey by default

imzs · web-flow · commit 79e700312a7b · 2025-12-24T10:00:01.000+08:00
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/RecallMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/RecallMessageProcessor.java
@@ -135,8 +135,8 @@ public MessageExtBrokerInner buildMessage(ChannelHandlerContext ctx, RecallMessa
         msgInner.setTags(RECALL_MESSAGE_TAG);
         msgInner.setTagsCode(RECALL_MESSAGE_TAG.hashCode());
         msgInner.setQueueId(0);
-        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TIMER_DEL_UNIQKEY,
-            TimerMessageStore.buildDeleteKey(handle.getTopic(), handle.getMessageId()));
+        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_TIMER_DEL_UNIQKEY, TimerMessageStore.buildDeleteKey(
+            handle.getTopic(), handle.getMessageId(), brokerController.getMessageStoreConfig().isAppendTopicForTimerDeleteKey()));
         MessageAccessor.putProperty(msgInner,
             MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, handle.getMessageId());
         MessageAccessor.putProperty(msgInner,
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/RecallMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/RecallMessageProcessorTest.java
@@ -96,7 +96,8 @@ public void init() throws IllegalAccessException, NoSuchFieldException {
     }
 
     @Test
-    public void testBuildMessage() {
+    public void testBuildMessage_withNamespace() {
+        when(messageStoreConfig.isAppendTopicForTimerDeleteKey()).thenReturn(true);
         String timestampStr = String.valueOf(System.currentTimeMillis());
         String id = "id";
         RecallMessageHandle.HandleV1 handle = new RecallMessageHandle.HandleV1(TOPIC, "brokerName", timestampStr, id);
@@ -110,6 +111,22 @@ public void testBuildMessage() {
         Assert.assertEquals(TOPIC + "+" + id, properties.get(MessageConst.PROPERTY_TIMER_DEL_UNIQKEY));
     }
 
+    @Test
+    public void testBuildMessage_withoutNamespace() {
+        when(messageStoreConfig.isAppendTopicForTimerDeleteKey()).thenReturn(false);
+        String timestampStr = String.valueOf(System.currentTimeMillis());
+        String id = "id";
+        RecallMessageHandle.HandleV1 handle = new RecallMessageHandle.HandleV1(TOPIC, "brokerName", timestampStr, id);
+        MessageExtBrokerInner msg =
+            recallMessageProcessor.buildMessage(handlerContext, new RecallMessageRequestHeader(), handle);
+
+        Assert.assertEquals(TOPIC, msg.getTopic());
+        Map<String, String> properties = MessageDecoder.string2messageProperties(msg.getPropertiesString());
+        Assert.assertEquals(timestampStr, properties.get(MessageConst.PROPERTY_TIMER_DELIVER_MS));
+        Assert.assertEquals(id, properties.get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
+        Assert.assertEquals(id, properties.get(MessageConst.PROPERTY_TIMER_DEL_UNIQKEY));
+    }
+
     @Test
     public void testHandlePutMessageResult() {
         MessageExt message = new MessageExt();
diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
@@ -533,6 +533,8 @@ public void setRocksdbCompressionType(String compressionType) {
 
     private boolean enableLogConsumeQueueRepeatedlyBuildWhenRecover = false;
 
+    private boolean appendTopicForTimerDeleteKey = false;
+
     public boolean isRocksdbCQDoubleWriteEnable() {
         return rocksdbCQDoubleWriteEnable;
     }
@@ -2236,4 +2238,12 @@ public int getSharedByteBufferNum() {
     public void setSharedByteBufferNum(int sharedByteBufferNum) {
         this.sharedByteBufferNum = sharedByteBufferNum;
     }
+
+    public boolean isAppendTopicForTimerDeleteKey() {
+        return appendTopicForTimerDeleteKey;
+    }
+
+    public void setAppendTopicForTimerDeleteKey(boolean appendTopicForTimerDeleteKey) {
+        this.appendTopicForTimerDeleteKey = appendTopicForTimerDeleteKey;
+    }
 }
diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
@@ -1741,7 +1741,7 @@ public void run() {
                                         isRound = false;
                                     }
                                     if (null != uniqueKey && tr.getDeleteList() != null && tr.getDeleteList().size() > 0
-                                        && tr.getDeleteList().contains(buildDeleteKey(getRealTopic(msgExt), uniqueKey))) {
+                                        && tr.getDeleteList().contains(buildDeleteKey(getRealTopic(msgExt), uniqueKey, storeConfig.isAppendTopicForTimerDeleteKey()))) {
                                         avoidDeleteLose.remove(uniqueKey);
                                         doRes = true;
                                         tr.idempotentRelease();
@@ -2074,9 +2074,9 @@ public TimerCheckpoint getTimerCheckpoint() {
         return timerCheckpoint;
     }
 
-    // identify a message by topic + uk, like query operation
-    public static String buildDeleteKey(String realTopic, String uniqueKey) {
-        return realTopic + "+" + uniqueKey;
+    // identify a message by topic or topic + uk(like query operation)
+    public static String buildDeleteKey(String realTopic, String uniqueKey, Boolean appendTopicForTimerDeleteKey) {
+        return appendTopicForTimerDeleteKey ? (realTopic + "+" + uniqueKey) : uniqueKey;
     }
 
     private void recallToTimeline(long delayTime, long offsetPy, int sizePy, MessageExt messageExt) {
diff --git a/store/src/test/java/org/apache/rocketmq/store/timer/TimerMessageStoreTest.java b/store/src/test/java/org/apache/rocketmq/store/timer/TimerMessageStoreTest.java
@@ -110,6 +110,7 @@ public void init() throws Exception {
         storeConfig.setFlushDiskType(FlushDiskType.ASYNC_FLUSH);
         storeConfig.setTimerInterceptDelayLevel(true);
         storeConfig.setTimerPrecisionMs(precisionMs);
+        storeConfig.setAppendTopicForTimerDeleteKey(false); // reset default value
 
         mockMessageStore = Mockito.mock(MessageStore.class);
         messageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("TimerTest",false), new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
@@ -358,7 +359,7 @@ public void testDeleteTimerMessage() throws Exception {
 
         MessageExtBrokerInner delMsg = buildMessage(delayMs, topic, false);
         transformTimerMessage(timerMessageStore,delMsg);
-        MessageAccessor.putProperty(delMsg, TimerMessageStore.TIMER_DELETE_UNIQUE_KEY, TimerMessageStore.buildDeleteKey(topic, uniqKey));
+        MessageAccessor.putProperty(delMsg, TimerMessageStore.TIMER_DELETE_UNIQUE_KEY, TimerMessageStore.buildDeleteKey(topic, uniqKey, false));
         delMsg.setPropertiesString(MessageDecoder.messageProperties2String(delMsg.getProperties()));
         assertEquals(PutMessageStatus.PUT_OK, messageStore.putMessage(delMsg).getPutMessageStatus());
 
@@ -375,6 +376,7 @@ public void testDeleteTimerMessage() throws Exception {
 
     @Test
     public void testDeleteTimerMessage_ukCollision() throws Exception {
+        storeConfig.setAppendTopicForTimerDeleteKey(true); // append topic as namespace
         String topic = "TimerTest_testDeleteTimerMessage";
         String collisionTopic = "TimerTest_testDeleteTimerMessage_collision";
 
@@ -397,13 +399,13 @@ public void testDeleteTimerMessage_ukCollision() throws Exception {
 
         MessageExtBrokerInner delMsg = buildMessage(delayMs, "whatever", false);
         transformTimerMessage(timerMessageStore, delMsg);
-        MessageAccessor.putProperty(delMsg, TimerMessageStore.TIMER_DELETE_UNIQUE_KEY, TimerMessageStore.buildDeleteKey(topic, firstUniqKey));
+        MessageAccessor.putProperty(delMsg, TimerMessageStore.TIMER_DELETE_UNIQUE_KEY, TimerMessageStore.buildDeleteKey(topic, firstUniqKey, true));
         delMsg.setPropertiesString(MessageDecoder.messageProperties2String(delMsg.getProperties()));
         assertEquals(PutMessageStatus.PUT_OK, messageStore.putMessage(delMsg).getPutMessageStatus());
 
         delMsg = buildMessage(delayMs, "whatever", false);
         transformTimerMessage(timerMessageStore, delMsg);
-        MessageAccessor.putProperty(delMsg, TimerMessageStore.TIMER_DELETE_UNIQUE_KEY, TimerMessageStore.buildDeleteKey(collisionTopic, secondUniqKey));
+        MessageAccessor.putProperty(delMsg, TimerMessageStore.TIMER_DELETE_UNIQUE_KEY, TimerMessageStore.buildDeleteKey(collisionTopic, secondUniqKey, true));
         delMsg.setPropertiesString(MessageDecoder.messageProperties2String(delMsg.getProperties()));
         assertEquals(PutMessageStatus.PUT_OK, messageStore.putMessage(delMsg).getPutMessageStatus());
 
diff --git a/test/src/test/java/org/apache/rocketmq/test/recall/SendAndRecallDelayMessageIT.java b/test/src/test/java/org/apache/rocketmq/test/recall/SendAndRecallDelayMessageIT.java
@@ -42,16 +42,34 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
+@RunWith(Parameterized.class)
 public class SendAndRecallDelayMessageIT extends BaseConf {
 
     private static String initTopic;
     private static String consumerGroup;
     private static RMQNormalProducer producer;
     private static RMQPopConsumer popConsumer;
 
+    private final boolean appendTopicForTimerDeleteKey;
+
+    public SendAndRecallDelayMessageIT(boolean appendTopicForTimerDeleteKey) {
+        this.appendTopicForTimerDeleteKey = appendTopicForTimerDeleteKey;
+    }
+
+    @Parameterized.Parameters
+    public static List<Object[]> params() {
+        List<Object[]> result = new ArrayList<>();
+        result.add(new Object[] {false});
+        result.add(new Object[] {true});
+        return result;
+    }
+
     @Before
     public void init() {
+        brokerController1.getMessageStoreConfig().setAppendTopicForTimerDeleteKey(appendTopicForTimerDeleteKey);
         initTopic = initTopic();
         consumerGroup = initConsumerGroup();
         producer = getProducer(NAMESRV_ADDR, initTopic);
@@ -126,6 +144,9 @@ public void testSendAndRecall() throws Exception {
 
     @Test
     public void testSendAndRecall_ukCollision() throws Exception {
+        if (!appendTopicForTimerDeleteKey) { // skip
+            return;
+        }
         int delaySecond = 5;
         String topic = MQRandomUtils.getRandomTopic();
         String collisionTopic = MQRandomUtils.getRandomTopic();

Original file line number	Diff line number	Diff line change
`@@ -533,6 +533,8 @@ public void setRocksdbCompressionType(String compressionType) {`
`533`	`533`
`534`	`534`	`private boolean enableLogConsumeQueueRepeatedlyBuildWhenRecover = false;`
`535`	`535`
	`536`	`+ private boolean appendTopicForTimerDeleteKey = false;`
	`537`	`+`
`536`	`538`	`public boolean isRocksdbCQDoubleWriteEnable() {`
`537`	`539`	`return rocksdbCQDoubleWriteEnable;`
`538`	`540`	`}`
`@@ -2236,4 +2238,12 @@ public int getSharedByteBufferNum() {`
`2236`	`2238`	`public void setSharedByteBufferNum(int sharedByteBufferNum) {`
`2237`	`2239`	`this.sharedByteBufferNum = sharedByteBufferNum;`
`2238`	`2240`	`}`
	`2241`	`+`
	`2242`	`+ public boolean isAppendTopicForTimerDeleteKey() {`
	`2243`	`+ return appendTopicForTimerDeleteKey;`
	`2244`	`+ }`
	`2245`	`+`
	`2246`	`+ public void setAppendTopicForTimerDeleteKey(boolean appendTopicForTimerDeleteKey) {`
	`2247`	`+ this.appendTopicForTimerDeleteKey = appendTopicForTimerDeleteKey;`
	`2248`	`+ }`
`2239`	`2249`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1741,7 +1741,7 @@ public void run() {`
`1741`	`1741`	`isRound = false;`
`1742`	`1742`	`}`
`1743`	`1743`	`if (null != uniqueKey && tr.getDeleteList() != null && tr.getDeleteList().size() > 0`
`1744`		`- && tr.getDeleteList().contains(buildDeleteKey(getRealTopic(msgExt), uniqueKey))) {`
	`1744`	`+ && tr.getDeleteList().contains(buildDeleteKey(getRealTopic(msgExt), uniqueKey, storeConfig.isAppendTopicForTimerDeleteKey()))) {`
`1745`	`1745`	`avoidDeleteLose.remove(uniqueKey);`
`1746`	`1746`	`doRes = true;`
`1747`	`1747`	`tr.idempotentRelease();`
`@@ -2074,9 +2074,9 @@ public TimerCheckpoint getTimerCheckpoint() {`
`2074`	`2074`	`return timerCheckpoint;`
`2075`	`2075`	`}`
`2076`	`2076`
`2077`		`- // identify a message by topic + uk, like query operation`
`2078`		`- public static String buildDeleteKey(String realTopic, String uniqueKey) {`
`2079`		`- return realTopic + "+" + uniqueKey;`
	`2077`	`+ // identify a message by topic or topic + uk(like query operation)`
	`2078`	`+ public static String buildDeleteKey(String realTopic, String uniqueKey, Boolean appendTopicForTimerDeleteKey) {`
	`2079`	`+ return appendTopicForTimerDeleteKey ? (realTopic + "+" + uniqueKey) : uniqueKey;`
`2080`	`2080`	`}`
`2081`	`2081`
`2082`	`2082`	`private void recallToTimeline(long delayTime, long offsetPy, int sizePy, MessageExt messageExt) {`