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 index f0367023ddd..48122d64c7d 100644 --- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java +++ b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java @@ -112,7 +112,6 @@ public class MessageStoreConfig { private long timerRocksDBPrecisionMs = 1000L; private double timerRocksDBRollMaxTps = 8000.0; private double timerRocksDBTimeExpiredMaxTps = 200000.0; - private int timerRocksDBRollIntervalHours = 1; private int timerRocksDBRollRangeHours = 2; private boolean timerRecallToTimeWheelEnable = true; private boolean timerRecallToTimelineEnable = true; @@ -2398,14 +2397,6 @@ public int getTimerReputServiceQueueCapacity() { return timerReputServiceQueueCapacity; } - public int getTimerRocksDBRollIntervalHours() { - return timerRocksDBRollIntervalHours; - } - - public void setTimerRocksDBRollIntervalHours(int timerRocksDBRollIntervalHours) { - this.timerRocksDBRollIntervalHours = timerRocksDBRollIntervalHours; - } - public int getTimerRocksDBRollRangeHours() { return timerRocksDBRollRangeHours; } diff --git a/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java b/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java index d55596a293c..029cc1883dc 100644 --- a/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java +++ b/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java @@ -73,9 +73,11 @@ public class MessageRocksDBStorage extends AbstractRocksDBStorage { private static final Set COMMON_CHECK_POINT_KEY_SET_FOR_TIMER = new HashSet<>(); public static final byte[] SYS_TOPIC_SCAN_OFFSET_CHECK_POINT = "sys_topic_scan_offset_checkpoint".getBytes(StandardCharsets.UTF_8); public static final byte[] TIMELINE_CHECK_POINT = "timeline_checkpoint".getBytes(StandardCharsets.UTF_8); + public static final byte[] TIMELINE_ROLL_CHECK_POINT = "timeline_roll_checkpoint".getBytes(StandardCharsets.UTF_8); static { COMMON_CHECK_POINT_KEY_SET_FOR_TIMER.add(SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); COMMON_CHECK_POINT_KEY_SET_FOR_TIMER.add(TIMELINE_CHECK_POINT); + COMMON_CHECK_POINT_KEY_SET_FOR_TIMER.add(TIMELINE_ROLL_CHECK_POINT); } private static final byte[] DELETE_VAL_FLAG = new byte[] {(byte)0xFF}; private static final int LAST_OFFSET_PY_LENGTH = LAST_OFFSET_PY.length; diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java index 740d5602b21..18ab52342f3 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java @@ -49,6 +49,8 @@ public class Timeline { private static final String DELETE_KEY_SPLIT = "+"; private static final int ORIGIN_CAPACITY = 100000; private static final int BATCH_SIZE = 1000, MAX_BATCH_SIZE_FROM_ROCKSDB = 8000; + private static final long ROLL_TRIGGER_EARLY_MS = 1000L; + private static final long ROLL_POLL_WHEN_NOT_DUE_MS = 1000L; private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; private volatile int state = INITIAL; private final AtomicLong commitOffset = new AtomicLong(0); @@ -373,37 +375,37 @@ public String getServiceName() { @Override public void run() { - log.info(this.getServiceName() + " service start"); + long checkpoint = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT); + if (checkpoint <= 0L) { + long now = System.currentTimeMillis(); + long forwardCheckpoint = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_CHECK_POINT); + int rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2; + checkpoint = (forwardCheckpoint > 0L ? Math.min(forwardCheckpoint, now) : now) + + TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()) - TimeUnit.HOURS.toMillis(rollRangeHour); + } + log.info(this.getServiceName() + " service start, checkpoint: {}", checkpoint); while (!this.isStopped()) { - int rollIntervalHour = 1; - int rollRangeHour = 2; try { - if (storeConfig.getTimerRocksDBRollIntervalHours() > 0) { - rollIntervalHour = storeConfig.getTimerRocksDBRollIntervalHours(); - } - if (storeConfig.getTimerRocksDBRollRangeHours() > 0) { - rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours(); - } - this.waitForRunning(TimeUnit.HOURS.toMillis(rollIntervalHour)); - if (stopped) { - log.info(this.getServiceName() + " service end"); - return; + long maxDelayMs = TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()); + long rangeMs = TimeUnit.HOURS.toMillis(storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2); + long triggerAt = checkpoint + rangeMs - maxDelayMs - ROLL_TRIGGER_EARLY_MS; + long now = System.currentTimeMillis(); + if (now < triggerAt) { + this.waitForRunning(ROLL_POLL_WHEN_NOT_DUE_MS); + continue; } - } catch (Exception e) { - logError.error("Timeline TimelineRollService wait error: {}", e.getMessage()); - } - long rollCheckpoint = System.currentTimeMillis(); - try { - log.info("Timeline TimelineRollService start roll rollCheckpoint: {}", rollCheckpoint); - while (!scanRecordsToQueue(rollCheckpoint + TimeUnit.HOURS.toMillis(rollRangeHour), - TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()), - timerMessageRocksDBStore.getRollMessageQueue())) { - logError.error("Timeline TimelineRollService scanRecordsToQueue error."); - Thread.sleep(200); + + log.info("Timeline TimelineRollService start roll checkpoint: {}, rangeMs: {}, triggerAt: {}, delayMs: {}", checkpoint, rangeMs, triggerAt, now - triggerAt); + if (!scanRecordsToQueue(checkpoint, rangeMs, timerMessageRocksDBStore.getRollMessageQueue())) { + logError.error("Timeline TimelineRollService scanRecordsToQueue error, checkpoint: {}", checkpoint); + this.waitForRunning(200L); + continue; } - log.info("Timeline TimelineRollService roll records success, lastRollTime: {}, rollCheckpoint: {}, cost: {}", rollCheckpoint, rollCheckpoint, System.currentTimeMillis() - rollCheckpoint); + checkpoint += rangeMs; + log.info("Timeline TimelineRollService roll records success, checkpoint: {}, cost: {}", checkpoint, System.currentTimeMillis() - now); } catch (Exception e) { logError.error("Timeline TimelineRollService failed error: {}", e.getMessage()); + this.waitForRunning(200L); } } log.info(this.getServiceName() + " service end"); 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 index c48e177c9d2..85a0d02ec91 100644 --- 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 @@ -231,8 +231,8 @@ private void initService() { this.expiredMessageQueue = new LinkedBlockingDeque<>(TIME_UP_CAPACITY); this.rollMessageQueue = new LinkedBlockingDeque<>(ROLL_CAPACITY); } - this.expiredMessageReputService = new TimerMessageReputService(expiredMessageQueue, storeConfig.getTimerRocksDBTimeExpiredMaxTps(), true); - this.rollMessageReputService = new TimerMessageReputService(rollMessageQueue, storeConfig.getTimerRocksDBRollMaxTps(), false); + this.expiredMessageReputService = new TimerMessageReputService(expiredMessageQueue, storeConfig.getTimerRocksDBTimeExpiredMaxTps(), MessageRocksDBStorage.TIMELINE_CHECK_POINT); + this.rollMessageReputService = new TimerMessageReputService(rollMessageQueue, storeConfig.getTimerRocksDBRollMaxTps(), MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT); this.timeline = new Timeline(messageStore, messageRocksDBStorage, this, timerMetrics); this.timerSysTopicScanService = new TimerSysTopicScanService(); } @@ -506,7 +506,7 @@ private class TimerMessageReputService extends ServiceThread { private final Logger log = TimerMessageRocksDBStore.log; private final BlockingQueue> queue; private final RateLimiter rateLimiter; - private final boolean writeCheckPoint; + private final byte[] checkPointKey; private final ExecutorService executor = ThreadUtils.newThreadPoolExecutor( storeConfig.getTimerReputServiceCorePoolSize(), @@ -518,10 +518,10 @@ private class TimerMessageReputService extends ServiceThread { new ThreadPoolExecutor.CallerRunsPolicy() ); - public TimerMessageReputService(BlockingQueue> queue, double maxTps, boolean writeCheckPoint) { + public TimerMessageReputService(BlockingQueue> queue, double maxTps, byte[] checkPointKey) { this.queue = queue; this.rateLimiter = RateLimiter.create(maxTps); - this.writeCheckPoint = writeCheckPoint; + this.checkPointKey = checkPointKey; } @Override @@ -545,9 +545,9 @@ public void run() { } countDownLatch.await(); log.info("TimerMessageReputService reput messages to commitlog, cost: {}, trs size: {}, checkPoint: {}", System.currentTimeMillis() - start, trs.size(), trs.get(trs.size() - 1).getCheckPoint()); - if (this.writeCheckPoint && !CollectionUtils.isEmpty(trs) && trs.get(trs.size() - 1).getCheckPoint() > 0L) { + if (null != this.checkPointKey && !CollectionUtils.isEmpty(trs) && trs.get(trs.size() - 1).getCheckPoint() > 0L) { log.info("TimerMessageReputService reput messages to commitlog, checkPoint: {}", trs.get(trs.size() - 1).getCheckPoint()); - messageRocksDBStorage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_CHECK_POINT, trs.get(trs.size() - 1).getCheckPoint()); + messageRocksDBStorage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, this.checkPointKey, trs.get(trs.size() - 1).getCheckPoint()); } } catch (Exception e) { logError.error("TimerMessageReputService error: {}", e.getMessage()); diff --git a/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java b/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java index d28ef19f54c..73c82bf674f 100644 --- a/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java @@ -29,10 +29,16 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMELINE_CHECK_POINT; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT; import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; public class MessageRocksDBStorageTest { + /** Fixed delay time so window assertions never depend on the wall clock. */ + private static final long FIXED_DELAY_TIME_BASE = 2000000000000L; + private static final long WINDOW = 3600000L; + private MessageRocksDBStorage storage; private String storePath; @@ -140,4 +146,58 @@ public void testDeleteThenUpdate() { Assert.assertEquals(0, recordCount); } + @Test + public void testWriteAndGetRollCheckpoint() { + Assert.assertEquals(0L, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + + long checkpoint = FIXED_DELAY_TIME_BASE; + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, checkpoint); + Assert.assertEquals(checkpoint, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + + long nextCheckpoint = checkpoint + WINDOW; + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, nextCheckpoint); + Assert.assertEquals(nextCheckpoint, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + } + + @Test + public void testRollCheckpointIsIndependentOfForwardCheckpoint() { + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_CHECK_POINT, FIXED_DELAY_TIME_BASE); + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, FIXED_DELAY_TIME_BASE + WINDOW); + + Assert.assertEquals(FIXED_DELAY_TIME_BASE, + storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_CHECK_POINT)); + Assert.assertEquals(FIXED_DELAY_TIME_BASE + WINDOW, + storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + } + + @Test + public void testScanAdjacentWindowsNoOverlap() { + long begin = FIXED_DELAY_TIME_BASE; + + writeTimerRecord(begin + 1, "roll-window-first", 11L, 111); + writeTimerRecord(begin + WINDOW, "roll-window-boundary", 22L, 222); + writeTimerRecord(begin + WINDOW + 1, "roll-window-second", 33L, 333); + + List firstWindow = storage.scanRecordsForTimer( + TIMER_COLUMN_FAMILY, begin, begin + WINDOW, 10, null); + Assert.assertNotNull(firstWindow); + Assert.assertEquals(1, firstWindow.size()); + Assert.assertEquals("roll-window-first", firstWindow.get(0).getUniqKey()); + + List secondWindow = storage.scanRecordsForTimer( + TIMER_COLUMN_FAMILY, begin + WINDOW, begin + 2 * WINDOW, 10, null); + Assert.assertNotNull(secondWindow); + Assert.assertEquals(2, secondWindow.size()); + Assert.assertEquals("roll-window-boundary", secondWindow.get(0).getUniqKey()); + Assert.assertEquals("roll-window-second", secondWindow.get(1).getUniqKey()); + } + + private void writeTimerRecord(long delayTime, String uniqKey, long offsetPy, int sizePy) { + TimerRocksDBRecord record = new TimerRocksDBRecord(delayTime, uniqKey, offsetPy, sizePy, 0L, null); + record.setActionFlag(TimerRocksDBRecord.TIMER_ROCKSDB_PUT); + List list = new ArrayList<>(); + list.add(record); + storage.writeRecordsForTimer(TIMER_COLUMN_FAMILY, list); + } + }