[ISSUE #9141] [RIP-75] Supports timer message on RocksDB

common/src/main/java/org/apache/rocketmq/common/message/MessageConst.java

@@ -95,6 +95,7 @@ public class MessageConst {
     public static final String PROPERTY_TIMER_ROLL_TIMES = "TIMER_ROLL_TIMES";
     public static final String PROPERTY_TIMER_OUT_MS = "TIMER_OUT_MS";
     public static final String PROPERTY_TIMER_DEL_UNIQKEY = "TIMER_DEL_UNIQKEY";
+    public static final String PROPERTY_TIMER_DEL_FLAG = "TIMER_DEL_FLAG";
     public static final String PROPERTY_TIMER_DELAY_LEVEL = "TIMER_DELAY_LEVEL";
     public static final String PROPERTY_TIMER_DELAY_MS = "TIMER_DELAY_MS";
     public static final String PROPERTY_CRC32 = "__CRC32#";
@@ -151,6 +152,7 @@ public class MessageConst {
         STRING_HASH_SET.add(PROPERTY_TIMER_ROLL_TIMES);
         STRING_HASH_SET.add(PROPERTY_TIMER_OUT_MS);
         STRING_HASH_SET.add(PROPERTY_TIMER_DEL_UNIQKEY);
+        STRING_HASH_SET.add(PROPERTY_TIMER_DEL_FLAG);
         STRING_HASH_SET.add(PROPERTY_TIMER_DELAY_LEVEL);
         STRING_HASH_SET.add(PROPERTY_BORN_HOST);
         STRING_HASH_SET.add(PROPERTY_BORN_TIMESTAMP);

store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java

@@ -472,6 +472,17 @@ public void setRocksdbCompressionType(String compressionType) {
      **/
     private boolean useABSLock = false;
 
+    /**
+     * Maximum number of messages to be read each time
+     * -1 : read all messages
+     */
+    private int readCountTimerOnRocksDB = -1;
+
+    /**
+     * When enabled, the scheduled message is using rocksdb
+     */
+    private boolean enableTimerMessageOnRocksDB = false;
+
     public boolean isRocksdbCQDoubleWriteEnable() {
         return rocksdbCQDoubleWriteEnable;
     }
@@ -1638,6 +1649,22 @@ public void setTimerPrecisionMs(int timerPrecisionMs) {
         this.timerPrecisionMs = candidates[candidates.length - 1];
     }
 
+    // visible for test
+    public void setTimerPrecision(int timerPrecisionMs) {
+        if (enableTimerMessageOnRocksDB) {
+            this.timerPrecisionMs = timerPrecisionMs;
+            return;
+        }
+        int[] candidates = {100, 200, 500, 1000};
+        for (int i = 1; i < candidates.length; i++) {
+            if (timerPrecisionMs < candidates[i]) {
+                this.timerPrecisionMs = candidates[i - 1];
+                return;
+            }
+        }
+        this.timerPrecisionMs = candidates[candidates.length - 1];
+    }
+
     public int getTimerRollWindowSlot() {
         return timerRollWindowSlot;
     }
@@ -1714,6 +1741,10 @@ public boolean isTimerStopDequeue() {
         return timerStopDequeue;
     }
 
+    public void setTimerStopDequeue(boolean timerStopDequeue) {
+        this.timerStopDequeue = timerStopDequeue;
+    }
+
     public int getTimerMetricSmallThreshold() {
         return timerMetricSmallThreshold;
     }
@@ -1950,4 +1981,20 @@ public void setUseABSLock(boolean useABSLock) {
     public boolean getUseABSLock() {
         return useABSLock;
     }
+
+    public void setReadCountTimerOnRocksDB(int readCountTimerOnRocksDB) {
+        this.readCountTimerOnRocksDB = readCountTimerOnRocksDB;
+    }
+
+    public int getReadCountTimerOnRocksDB() {
+        return readCountTimerOnRocksDB;
+    }
+
+    public void setEnableTimerMessageOnRocksDB(boolean enableTimerMessageOnRocksDB) {
+        this.enableTimerMessageOnRocksDB = enableTimerMessageOnRocksDB;
+    }
+
+    public boolean getEnableTimerMessageOnRocksDB() {
+        return enableTimerMessageOnRocksDB;
+    }
 }

store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java

@@ -182,6 +182,108 @@ public static ColumnFamilyOptions createPopCFOptions() {
             .setOptimizeFiltersForHits(true);
     }
 
+    public static ColumnFamilyOptions createTimerCFOptions() {
+        BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig()
+                .setFormatVersion(5)
+                .setIndexType(IndexType.kBinarySearch)
+                .setDataBlockIndexType(DataBlockIndexType.kDataBlockBinaryAndHash)
+                .setDataBlockHashTableUtilRatio(0.75)
+                .setBlockSize(128 * SizeUnit.KB)
+                .setMetadataBlockSize(4 * SizeUnit.KB)
+                .setFilterPolicy(new BloomFilter(16, false))
+                .setCacheIndexAndFilterBlocks(false)
+                .setCacheIndexAndFilterBlocksWithHighPriority(true)
+                .setPinL0FilterAndIndexBlocksInCache(false)
+                .setPinTopLevelIndexAndFilter(true)
+                .setBlockCache(new LRUCache(1024 * SizeUnit.MB, 8, false))
+                .setWholeKeyFiltering(true);
+
+        CompactionOptionsUniversal compactionOption = new CompactionOptionsUniversal()
+                .setSizeRatio(100)
+                .setMaxSizeAmplificationPercent(25)
+                .setAllowTrivialMove(true)
+                .setMinMergeWidth(2)
+                .setMaxMergeWidth(Integer.MAX_VALUE)
+                .setStopStyle(CompactionStopStyle.CompactionStopStyleTotalSize)
+                .setCompressionSizePercent(-1);
+
+        //noinspection resource
+        return new ColumnFamilyOptions()
+                .setMaxWriteBufferNumber(6)
+                .setWriteBufferSize(128 * SizeUnit.MB)
+                .setMinWriteBufferNumberToMerge(1)
+                .setTableFormatConfig(blockBasedTableConfig)
+                .setMemTableConfig(new SkipListMemTableConfig())
+                .setCompressionType(CompressionType.NO_COMPRESSION)
+                .setBottommostCompressionType(CompressionType.NO_COMPRESSION)
+                .setNumLevels(7)
+                .setCompactionPriority(CompactionPriority.MinOverlappingRatio)
+                .setCompactionStyle(CompactionStyle.UNIVERSAL)
+                .setCompactionOptionsUniversal(compactionOption)
+                .setMaxCompactionBytes(100 * SizeUnit.GB)
+                .setSoftPendingCompactionBytesLimit(100 * SizeUnit.GB)
+                .setHardPendingCompactionBytesLimit(256 * SizeUnit.GB)
+                .setLevel0FileNumCompactionTrigger(2)
+                .setLevel0SlowdownWritesTrigger(8)
+                .setLevel0StopWritesTrigger(10)
+                .setTargetFileSizeBase(256 * SizeUnit.MB)
+                .setTargetFileSizeMultiplier(2)
+                .setMergeOperator(new StringAppendOperator())
+                .setReportBgIoStats(true)
+                .setOptimizeFiltersForHits(true);
+    }
+
+    public static ColumnFamilyOptions createTimerMetricCFOptions() {
+        BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig()
+                .setFormatVersion(5)
+                .setIndexType(IndexType.kBinarySearch)
+                .setDataBlockIndexType(DataBlockIndexType.kDataBlockBinaryAndHash)
+                .setDataBlockHashTableUtilRatio(0.75)
+                .setBlockSize(4 * SizeUnit.KB)
+                .setMetadataBlockSize(4 * SizeUnit.KB)
+                .setFilterPolicy(new BloomFilter(16, false))
+                .setCacheIndexAndFilterBlocks(false)
+                .setCacheIndexAndFilterBlocksWithHighPriority(true)
+                .setPinL0FilterAndIndexBlocksInCache(false)
+                .setPinTopLevelIndexAndFilter(true)
+                .setBlockCache(new LRUCache(64 * SizeUnit.MB, 8, false))
+                .setWholeKeyFiltering(true);
+
+        CompactionOptionsUniversal compactionOption = new CompactionOptionsUniversal()
+                .setSizeRatio(100)
+                .setMaxSizeAmplificationPercent(25)
+                .setAllowTrivialMove(true)
+                .setMinMergeWidth(2)
+                .setMaxMergeWidth(Integer.MAX_VALUE)
+                .setStopStyle(CompactionStopStyle.CompactionStopStyleTotalSize)
+                .setCompressionSizePercent(-1);
+
+        //noinspection resource
+        return new ColumnFamilyOptions()
+                .setMaxWriteBufferNumber(4)
+                .setWriteBufferSize(64 * SizeUnit.MB)
+                .setMinWriteBufferNumberToMerge(1)
+                .setTableFormatConfig(blockBasedTableConfig)
+                .setMemTableConfig(new SkipListMemTableConfig())
+                .setCompressionType(CompressionType.NO_COMPRESSION)
+                .setBottommostCompressionType(CompressionType.NO_COMPRESSION)
+                .setNumLevels(7)
+                .setCompactionPriority(CompactionPriority.MinOverlappingRatio)
+                .setCompactionStyle(CompactionStyle.UNIVERSAL)
+                .setCompactionOptionsUniversal(compactionOption)
+                .setMaxCompactionBytes(100 * SizeUnit.GB)
+                .setSoftPendingCompactionBytesLimit(100 * SizeUnit.GB)
+                .setHardPendingCompactionBytesLimit(256 * SizeUnit.GB)
+                .setLevel0FileNumCompactionTrigger(2)
+                .setLevel0SlowdownWritesTrigger(8)
+                .setLevel0StopWritesTrigger(10)
+                .setTargetFileSizeBase(128 * SizeUnit.MB)
+                .setTargetFileSizeMultiplier(2)
+                .setMergeOperator(new StringAppendOperator())
+                .setReportBgIoStats(true)
+                .setOptimizeFiltersForHits(true);
+    }
+
     /**
      * Create a rocksdb db options, the user must take care to close it after closing db.
      * @return

store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java

@@ -71,6 +71,7 @@
 import org.apache.rocketmq.store.queue.CqUnit;
 import org.apache.rocketmq.store.queue.ReferredIterator;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
+import org.apache.rocketmq.store.timer.rocksdb.TimerMessageRocksDBStore;
 import org.apache.rocketmq.store.util.PerfCounter;
 
 public class TimerMessageStore {
@@ -158,6 +159,7 @@ public class TimerMessageStore {
     protected volatile boolean shouldRunningDequeue;
     private final BrokerStatsManager brokerStatsManager;
     private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook;
+    private TimerMessageRocksDBStore timerRocksDBStore;
 
     public TimerMessageStore(final MessageStore messageStore, final MessageStoreConfig storeConfig,
         TimerCheckpoint timerCheckpoint, TimerMetrics timerMetrics,
@@ -212,6 +214,7 @@ protected ByteBuffer initialValue() {
             dequeuePutQueue = new LinkedBlockingDeque<>(DEFAULT_CAPACITY);
         }
         this.brokerStatsManager = brokerStatsManager;
+        this.timerRocksDBStore = new TimerMessageRocksDBStore(messageStore, storeConfig, timerMetrics, brokerStatsManager);
     }
 
     public void initService() {
@@ -238,6 +241,7 @@ public boolean load() {
         this.initService();
         boolean load = timerLog.load();
         load = load && this.timerMetrics.load();
+        load = load && timerRocksDBStore.load();
         recover();
         calcTimerDistribution();
         return load;
@@ -464,6 +468,7 @@ public static boolean isMagicOK(int magic) {
     }
 
     public void start() {
+        timerRocksDBStore.start();
         this.shouldStartTime = storeConfig.getDisappearTimeAfterStart() + System.currentTimeMillis();
         maybeMoveWriteTime();
         enqueueGetService.start();
@@ -504,7 +509,11 @@ public void run() {
                         long curr = System.currentTimeMillis();
                         if (curr - lastTimeOfCheckMetrics > 70 * 60 * 1000) {
                             lastTimeOfCheckMetrics = curr;
-                            checkAndReviseMetrics();
+                            if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+                                timerRocksDBStore.checkAndReviseMetrics();
+                            } else {
+                                checkAndReviseMetrics();
+                            }
                             LOGGER.info("[CheckAndReviseMetrics]Timer do check timer metrics cost {} ms",
                                 System.currentTimeMillis() - curr);
                         }
@@ -552,6 +561,7 @@ public void shutdown() {
         timerWheel.shutdown(false);
 
         this.scheduler.shutdown();
+        this.timerRocksDBStore.shutdown();
         UtilAll.cleanBuffer(this.bufferLocal.get());
         this.bufferLocal.remove();
     }
@@ -653,7 +663,7 @@ public void holdMomentForUnknownError() {
     }
 
     public boolean enqueue(int queueId) {
-        if (storeConfig.isTimerStopEnqueue()) {
+        if (storeConfig.isTimerStopEnqueue() || storeConfig.getEnableTimerMessageOnRocksDB()) {
             return false;
         }
         if (!isRunningEnqueue()) {
@@ -1742,14 +1752,23 @@ public void run() {
     }
 
     public long getAllCongestNum() {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getAllCongestNum();
+        }
         return timerWheel.getAllNum(currReadTimeMs);
     }
 
     public long getCongestNum(long deliverTimeMs) {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getCongestNum(deliverTimeMs);
+        }
         return timerWheel.getNum(deliverTimeMs);
     }
 
     public boolean isReject(long deliverTimeMs) {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.isReject(deliverTimeMs);
+        }
         long congestNum = timerWheel.getNum(deliverTimeMs);
         if (congestNum <= storeConfig.getTimerCongestNumEachSlot()) {
             return false;
@@ -1764,13 +1783,19 @@ public boolean isReject(long deliverTimeMs) {
     }
 
     public long getEnqueueBehindMessages() {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getEnqueueBehindMessages();
+        }
         long tmpQueueOffset = currQueueOffset;
         ConsumeQueueInterface cq = messageStore.getConsumeQueue(TIMER_TOPIC, 0);
         long maxOffsetInQueue = cq == null ? 0 : cq.getMaxOffsetInQueue();
         return maxOffsetInQueue - tmpQueueOffset;
     }
 
     public long getEnqueueBehindMillis() {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getEnqueueBehindMillis();
+        }
         if (System.currentTimeMillis() - lastEnqueueButExpiredTime < 2000) {
             return System.currentTimeMillis() - lastEnqueueButExpiredStoreTime;
         }
@@ -1782,10 +1807,13 @@ public long getEnqueueBehind() {
     }
 
     public long getDequeueBehindMessages() {
-        return timerWheel.getAllNum(currReadTimeMs);
+        return getAllCongestNum();
     }
 
     public long getDequeueBehindMillis() {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getDequeueBehindMillis();
+        }
         return System.currentTimeMillis() - currReadTimeMs;
     }
 
@@ -1794,10 +1822,16 @@ public long getDequeueBehind() {
     }
 
     public float getEnqueueTps() {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getEnqueueTps();
+        }
         return perfCounterTicks.getCounter(ENQUEUE_PUT).getLastTps();
     }
 
     public float getDequeueTps() {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return timerRocksDBStore.getDequeueTps();
+        }
         return perfCounterTicks.getCounter("dequeue_put").getLastTps();
     }
 
@@ -1930,4 +1964,12 @@ public TimerCheckpoint getTimerCheckpoint() {
     public static String buildDeleteKey(String realTopic, String uniqueKey) {
         return realTopic + "+" + uniqueKey;
     }
+
+    public static String extractUniqueKey(String deleteKey) {
+        int separatorIndex = deleteKey.indexOf('+');
+        if (separatorIndex == -1) {
+            throw new IllegalArgumentException("Invalid deleteKey format");
+        }
+        return deleteKey.substring(separatorIndex + 1);
+    }
 }