Move the new class to TimerMessageStore

apache · Jan 20, 2025 · 3c4e922 · 3c4e922
1 parent 42d432a
commit 3c4e922
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 2 deletions.
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
@@ -1742,7 +1742,7 @@ public long getCongestNum(long deliverTimeMs) {
 
     public boolean isReject(long deliverTimeMs) {
         if (storeConfig.getEnableTimerMessageOnRocksDB()) {
-            return true;
+            return false;
         }
         long congestNum = timerWheel.getNum(deliverTimeMs);
         if (congestNum <= storeConfig.getTimerCongestNumEachSlot()) {

diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java
@@ -239,7 +239,14 @@ private String getServiceThreadName() {
     }
 
     private byte[] getColumnFamily(int flag) {
-        TimerMessageRecord.Flag tag = TimerMessageRecord.Flag.valueOf(String.valueOf(flag));
+        TimerMessageRecord.Flag tag;
+        switch (flag) {
+            case 1 : tag = TimerMessageRecord.Flag.TRANSACTION;
+                break;
+            case 2 : tag = TimerMessageRecord.Flag.POP;
+                break;
+            default : tag = TimerMessageRecord.Flag.DEFAULT;
+        }
         if (TimerMessageRecord.Flag.TRANSACTION == tag) {
             return TRANSACTION_COLUMN_FAMILY;
         } else if (tag == TimerMessageRecord.Flag.POP) {

diff --git a/...e/src/test/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStoreTest.java b/...e/src/test/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStoreTest.java
@@ -80,6 +80,8 @@ public void setUp() throws Exception {
         String baseDir = StoreTestUtils.createBaseDir();
         this.baseDirs.add(baseDir);
         storeConfig = new MessageStoreConfig();
+        storeConfig.isEnableRocksDBStore();
+        storeConfig.setEnableTimerMessageOnRocksDB(true);
         storeConfig.setStorePathRootDir(baseDir);
         storeConfig.setStorePathCommitLog(baseDir + File.separator + "commitlog");
         messageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("TimerTest",
@@ -180,6 +182,7 @@ private PutMessageResult transformTimerMessage(TimerMessageRocksDBStore timerMes
 
     @Test
     public void testDoNormalTimer() throws Exception {
+        Assume.assumeFalse(MixAll.isWindows());
         String topic = "TimerTest_testPutTimerMessage";
 
         final TimerMessageRocksDBStore timerMessageStore = createTimerMessageRocksDBStore(null);