opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-10-24 12:11:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #10972] Encode timer message propertiesString after internal properties are cleared (#10974)

Browse Source 
Co-authored-by: wangjiahua.wjh <wangjiahua.wjh@alibaba-inc.com>
This commit is contained in:
Jiahua Wang
2026-09-03 15:54:43 +08:00
committed by GitHub
co-authored by wangjiahua.wjh
parent bee586bcd0
commit e533b663fb
2 changed files with 28 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
+3 -1
View File
@@ -1251,7 +1251,6 @@ public class TimerMessageStore {
long tagsCodeValue =
MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
msgInner.setTagsCode(tagsCodeValue);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
msgInner.setSysFlag(msgExt.getSysFlag());
msgInner.setBornTimestamp(msgExt.getBornTimestamp());
@@ -1270,6 +1269,9 @@ public class TimerMessageStore {
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC);
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID);
}
// Encode after the properties are finalized so that the wire data stays consistent
// with the property map, aligning with TimerMessageRocksDBStore#convertMessage.
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
return msgInner;
}
store/src/test/java/org/apache/rocketmq/store/timer/TimerMessageStoreTest.java
+25
View File
@@ -179,6 +179,31 @@ public class TimerMessageStoreTest {
return null;
}
@Test
public void testConvertMessagePropertiesStringMatchesProperties() throws Exception {
final TimerMessageStore timerMessageStore = createTimerMessageStore(null, true);
MessageExtBrokerInner msgExt = buildMessage(3000, "TimerTest_testConvertMessage", false);
MessageAccessor.putProperty(msgExt, MessageConst.PROPERTY_REAL_TOPIC, msgExt.getTopic());
MessageAccessor.putProperty(msgExt, MessageConst.PROPERTY_REAL_QUEUE_ID, "0");
msgExt.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));
msgExt.setTopic(TimerMessageStore.TIMER_TOPIC);
// delivered message: internal properties are cleared from both the map and the wire data
MessageExtBrokerInner delivered = timerMessageStore.convertMessage(msgExt, false);
assertEquals("TimerTest_testConvertMessage", delivered.getTopic());
assertFalse(delivered.getPropertiesString().contains(MessageConst.PROPERTY_REAL_TOPIC));
assertFalse(delivered.getPropertiesString().contains(MessageConst.PROPERTY_REAL_QUEUE_ID));
assertEquals(MessageDecoder.messageProperties2String(delivered.getProperties()), delivered.getPropertiesString());
// rolled message: keeps REAL_TOPIC and stays consistent between the map and the wire data
MessageExtBrokerInner rolled = timerMessageStore.convertMessage(msgExt, true);
assertEquals(TimerMessageStore.TIMER_TOPIC, rolled.getTopic());
assertTrue(rolled.getPropertiesString().contains(MessageConst.PROPERTY_REAL_TOPIC));
assertTrue(rolled.getPropertiesString().contains(MessageConst.PROPERTY_REAL_QUEUE_ID));
assertEquals(MessageDecoder.messageProperties2String(rolled.getProperties()), rolled.getPropertiesString());
}
@Test
public void testPutTimerMessage() throws Exception {
Assume.assumeFalse(MixAll.isWindows());