[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;
     }
@@ -1714,6 +1725,10 @@ public boolean isTimerStopDequeue() {
         return timerStopDequeue;
     }
 
+    public void setTimerStopDequeue(boolean timerStopDequeue) {
+        this.timerStopDequeue = timerStopDequeue;
+    }
+
     public int getTimerMetricSmallThreshold() {
         return timerMetricSmallThreshold;
     }
@@ -1950,4 +1965,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();
@@ -552,6 +557,7 @@ public void shutdown() {
         timerWheel.shutdown(false);
 
         this.scheduler.shutdown();
+        this.timerRocksDBStore.shutdown();
         UtilAll.cleanBuffer(this.bufferLocal.get());
         this.bufferLocal.remove();
     }
@@ -653,7 +659,7 @@ public void holdMomentForUnknownError() {
     }
 
     public boolean enqueue(int queueId) {
-        if (storeConfig.isTimerStopEnqueue()) {
+        if (storeConfig.isTimerStopEnqueue() || storeConfig.getEnableTimerMessageOnRocksDB()) {
             return false;
         }
         if (!isRunningEnqueue()) {
@@ -1735,6 +1741,9 @@ public long getCongestNum(long deliverTimeMs) {
     }
 
     public boolean isReject(long deliverTimeMs) {
+        if (storeConfig.getEnableTimerMessageOnRocksDB()) {
+            return false;
+        }
         long congestNum = timerWheel.getNum(deliverTimeMs);
         if (congestNum <= storeConfig.getTimerCongestNumEachSlot()) {
             return false;
@@ -1915,4 +1924,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);
+    }
 }

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

@@ -80,6 +80,11 @@ public long updateDistPair(int period, int value) {
         return distPair.getCount().addAndGet(value);
     }
 
+    public void resetDistPair(int period, int value) {
+        Metric distPair = getDistPair(period);
+        distPair.getCount().set(value);
+    }
+
     public long addAndGet(MessageExt msg, int value) {
         String topic = msg.getProperty(MessageConst.PROPERTY_REAL_TOPIC);
         Metric pair = getTopicPair(topic);

store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageKVStore.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.store.timer.rocksdb;
+
+import java.util.List;
+
+public interface TimerMessageKVStore {
+
+    /**
+     * Start the timer message kv store.
+     */
+    boolean start();
+
+    /**
+     * Shutdown the timer message kv store.
+     */
+    boolean shutdown();
+
+    /**
+     * Get the file path of the timer message kv store.
+     */
+    String getFilePath();
+
+    /**
+     * Write the timer message records to the timer message kv store.
+     * @param columnFamily the column family of the timer message kv store.
+     * @param consumerRecordList the list of timer message records to be written.
+     * @param offset the cq offset of the timer message records to be written.
+     * @param timestamp the key of the timer message metric column family.
+     */
+    void writeAssignRecords(byte[] columnFamily, List<TimerMessageRecord> consumerRecordList, long offset, int timestamp);
+
+    /**
+     * Delete the timer message records from the timer message kv store.
+     * @param columnFamily the column family of the timer message kv store.
+     * @param consumerRecordList the list of timer message records to be deleted.
+     * @param timestamp the key of the timer message metric column family.
+     */
+    void deleteAssignRecords(byte[] columnFamily, List<TimerMessageRecord> consumerRecordList, int timestamp, long offset);
+
+    /**
+     * Scan the timer message records from the timer message kv store.
+     * @param columnFamily the column family of the timer message kv store.
+     * @param lowerTime the lower time of the timer message records to be scanned.
+     * @param upperTime the upper time of the timer message records to be scanned.
+     * @return the list of timer message records.
+     */
+    List<TimerMessageRecord> scanRecords(byte[] columnFamily, long lowerTime, long upperTime, long timestamp);
+
+    /**
+     * Get the commit offset of the timer message kv store from cq.
+     * @return the commit offset of the timer message kv store.
+     */
+    long getCommitOffset();
+
+    /**
+     * Get the metric size of the timer message kv store.
+     * @param lowerTime the lower time of the timer message records to be scanned.
+     * @param upperTime the upper time of the timer message records to be scanned.
+     * @return sum.
+     */
+    int getMetricSize(int lowerTime, int upperTime);
+
+    /**
+     * Get the checkpoint of the timer message kv store.
+     * @param columnFamily the column family of the timer message kv store.
+     * @return the checkpoint of the timer message kv store.
+     */
+    long getCheckpoint(byte[] columnFamily);
+}