[RIP-82] Implement Timer message, transaction message, and index based on RocksDB #9787

zk-drizzle · 2025-10-29T01:56:13Z

RIP-82 Implement Timer message, transaction message, and index based on RocksDB

Fixes #9786

Brief Description

Implement Timer message, transaction message, and index based on RocksDB

How Did You Test This Change?

Change-Id: I7b5782e8f1a9548851069c8c239c75868ae64400

Change-Id: Ica01fbd08a2d60e9a0f4ab54b41ae4768cc4fa5d

Change-Id: I764ce6cea3159b025bf1e6e65b60e20514cecb4f

Change-Id: I0cc735896e1266934293ec06cb4f07746a1894a1

Change-Id: I6538b6a60631c62e162bf511ae20185a07b72508

Change-Id: I03f83b9e8199f3595ca28471df4a2e94671fbaae

Change-Id: I5b19b671fa0b40cd2aba63f1d0eb654b92775cb4

Change-Id: I22788d112b7f214310719e2d3ebad839ef818e67

Change-Id: I7da89fe635b58ac5026095a8e484fa46f68521e7

Change-Id: I5eade6c31803b8cf05c1a4bb83c226d57f4980c7

Change-Id: Id67f600fad1a876926cd505dba047cc82011160c

Change-Id: Ia4b0e2a21aa5e12570a80713432fd48ddfb210e6

Change-Id: I65a43a612f09280cb61d2a23b3324d198c6a71de

Change-Id: Idf8080c7b17c25e14fe34ef6bad1e1150dab58d6

Change-Id: I904f0120a728b4eb87226159f9c0cde3d18ef26a

Change-Id: I21653d6bcad905daa8f10d12298d1d27535af9d9

Change-Id: I712b9cde9a18c730fd020ea76f05560e594a9edd

codecov-commenter · 2025-10-29T07:38:53Z

Codecov Report

❌ Patch coverage is 12.91667% with 2090 lines in your changes missing coverage. Please review.
✅ Project coverage is 47.40%. Comparing base (bea086f) to head (9fa994b).

Files with missing lines	Patch %	Lines
.../rocketmq/store/rocksdb/MessageRocksDBStorage.java	14.42%	356 Missing ⚠️
.../store/timer/rocksdb/TimerMessageRocksDBStore.java	0.00%	321 Missing ⚠️
.../apache/rocketmq/store/timer/rocksdb/Timeline.java	0.00%	253 Missing ⚠️
...mq/store/transaction/TransMessageRocksDBStore.java	0.00%	193 Missing ⚠️
...ocketmq/store/index/rocksdb/IndexRocksDBStore.java	8.98%	160 Missing and 2 partials ⚠️
...on/rocksdb/TransactionalMessageRocksDBService.java	0.00%	145 Missing ⚠️
...rocketmq/store/transaction/TransRocksDBRecord.java	0.00%	72 Missing ⚠️
...cketmq/store/timer/rocksdb/TimerRocksDBRecord.java	0.00%	64 Missing ⚠️
...org/apache/rocketmq/store/DefaultMessageStore.java	16.00%	58 Missing and 5 partials ⚠️
...ocketmq/broker/processor/AdminBrokerProcessor.java	0.00%	50 Missing ⚠️
... and 26 more

Additional details and impacted files

@@              Coverage Diff              @@
##             develop    #9787      +/-   ##
=============================================
- Coverage      48.40%   47.40%   -1.00%     
+ Complexity     12253    12252       -1     
=============================================
  Files           1314     1324      +10     
  Lines          93668    96023    +2355     
  Branches       12011    12371     +360     
=============================================
+ Hits           45340    45520     +180     
- Misses         42737    44853    +2116     
- Partials        5591     5650      +59

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

RongtongJin · 2025-10-29T11:19:46Z

pom.xml


    <properties>
-        <revision>5.3.4-SNAPSHOT</revision>
+        <revision>5.3.6-rocksdb-SNAPSHOT</revision>


The version number should not be changed here.

RongtongJin · 2025-10-30T05:53:32Z

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

+                    db.flush(flushOptions, timerCFHandle);
+                    log.info("MessageRocksDBStorage flush timer wal success");
+                } catch (Exception e) {
+                    logError.error("MessageRocksDBStorage flush timer wal failed, error: {}", e.getMessage());


Directly printing the exception is preferable, as printing e.getMessage() may lose important stack trace information, hindering troubleshooting.

majialoong · 2025-10-30T09:34:09Z

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

+            } else {
+                response.setCode(ResponseCode.SYSTEM_ERROR);
+                response.setRemark("switch timer engine error");
+                LOGGER.info("switchTimerEngine error");


Hello, @zk-drizzle . I'm just curious why all the logs here are INFO level, regardless of whether the restart succeeds or fails.

Shouldn't WARN or ERROR level logs be used to provide some warning information to the user when the restart fails?

majialoong · 2025-10-30T13:29:01Z

.../java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java

+        while (true) {
+            try {
+                List<TransRocksDBRecord> trs = messageRocksDBStorage.scanRecordsForTrans(TRANS_COLUMN_FAMILY, MAX_BATCH_SIZE_FROM_ROCKSDB, lastKey);
+                if (null == trs || CollectionUtils.isEmpty(trs)) {


The null == trs check is redundant.

Suggested change

if (null == trs || CollectionUtils.isEmpty(trs)) {

if (CollectionUtils.isEmpty(trs)) {

majialoong · 2025-10-30T13:52:09Z

.../java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java

+    }
+
+    public void shutdown() {
+        if (this.state != RUNNING || this.state == SHUTDOWN) {


Hello, @zk-drizzle . Is the check this.state == SHUTDOWN redundant? When state equals SHUTDOWN, the condition this.state != RUNNING is already satisfied.

majialoong

Thanks for @zk-drizzle PR, this is an exciting feature. I've left some comments.

majialoong · 2025-10-30T14:10:20Z

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

+    }
+
+    public void shutdown() {
+        if (this.state != RUNNING || this.state == SHUTDOWN) {


majialoong · 2025-10-30T14:40:03Z

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

+    }
+
+    public long getCommitOffsetInRocksDB() {
+        if (null == messageRocksDBStorage || !storeConfig.isTransRocksDBEnable()) {


Should this be isTimerRocksDBEnable?

majialoong · 2025-10-30T14:43:46Z

.../java/org/apache/rocketmq/broker/transaction/rocksdb/TransactionalMessageRocksDBService.java

+
+    private void checkTransRecordsStatus(List<TransRocksDBRecord> trs) {
+        if (CollectionUtils.isEmpty(trs)) {
+            log.error("TransactionalMessageRocksDBService checkTransRecordsStatus, trs is empty");


The checkTransRecordsStatus method is currently only called within the checkTransStatus method, and since the checkTransStatus method already checks for empty values, this method checks again for redundancy. Furthermore, the log level here is ERROR, while checkTransStatus uses INFO, creating inconsistencies.

majialoong · 2025-10-30T14:45:23Z

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

+            this.storeConfig.setTimerStopEnqueue(true);
+            if (this.state == RUNNING && !this.storeConfig.isTimerRocksDBStopScan()) {
+                log.info("restart TimerMessageRocksDBStore has been running");
+                return true;


I'm curious, if the enqueue operation was stopped using setTimerStopEnqueue(true) above, will there be any issues if we simply return here?

majialoong · 2025-10-30T14:45:30Z

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

+                long commitOffsetRocksDB = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.SYS_TOPIC_SCAN_OFFSET_CHECK_POINT);
+                long maxCommitOffset = Math.max(commitOffsetFile, commitOffsetRocksDB);
+                this.readOffset.set(maxCommitOffset);
+                log.info("restart TimerMessageRocksDBStore has benn recover running, commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, readOffset.get());


maybe is has been recovered to running

majialoong · 2025-10-30T14:46:26Z

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

+                this.readOffset.set(maxCommitOffset);
+                log.info("restart TimerMessageRocksDBStore has benn recover running, commitOffsetFile: {}, commitOffsetRocksDB: {}, readOffset: {}", commitOffsetFile, commitOffsetRocksDB, readOffset.get());
+            } else {
+                this.load();


If load fails here, should we continue with the subsequent start?

drizzle.zk and others added 30 commits November 14, 2024 16:00

add isWakeCommitWhenPutMessage for AIO

cac4f78

Merge branch 'apache:develop' into develop

38f03d0

Merge branch 'apache:develop' into develop

63d0eb6

Implemented Timer message, Transaction Message, Index based on rocksdb

e75c906

add timeline roll interval can be config

09f7671

optimize the code

3d4aa00

optimize the code

0d0efeb

add sone logic for recover commitlog

63cd3bf

optimize the code

1184544

update

343adc8

update

7c5b86f

optimize the code

fb3fdf0

update

35bb4f5

update

3945e81

optimize the code

ca36da3

optimize the code

e9899ca

update

679b4d1

update

361d2d9

update

653ba0a

update

dadb1ac

update

900e29f

update

c290e67

update

04738f9

update

2ff33d0

update

18eb884

update

41e5782

update

de75d1b

update

1f1727b

Merge branch 'apache:develop' into develop

4aada4b

Implement Timer message, transaction message, and index based on RocksDB

cd4a00e

Change-Id: I7b5782e8f1a9548851069c8c239c75868ae64400

zhouli11 and others added 16 commits October 23, 2025 11:06

update

b606d55

Change-Id: Ica01fbd08a2d60e9a0f4ab54b41ae4768cc4fa5d

update

3e37b98

Change-Id: I764ce6cea3159b025bf1e6e65b60e20514cecb4f

update

be6cac4

Change-Id: I0cc735896e1266934293ec06cb4f07746a1894a1

update

b9c9022

Change-Id: I6538b6a60631c62e162bf511ae20185a07b72508

update

6ee5481

Change-Id: I03f83b9e8199f3595ca28471df4a2e94671fbaae

update

c67aae4

Change-Id: I5b19b671fa0b40cd2aba63f1d0eb654b92775cb4

update the code

2872ab4

Change-Id: I22788d112b7f214310719e2d3ebad839ef818e67

update

1739ba1

Change-Id: I7da89fe635b58ac5026095a8e484fa46f68521e7

update the code

599777e

Change-Id: I5eade6c31803b8cf05c1a4bb83c226d57f4980c7

update

62e9b9f

Change-Id: Id67f600fad1a876926cd505dba047cc82011160c

update

e077a1a

Change-Id: Ia4b0e2a21aa5e12570a80713432fd48ddfb210e6

Merge branch 'apache:develop' into develop

67f4ef0

update code

98757c6

Change-Id: I65a43a612f09280cb61d2a23b3324d198c6a71de

optimize the code

5224bfc

Change-Id: Idf8080c7b17c25e14fe34ef6bad1e1150dab58d6

Merge branch 'apache:develop' into develop

fd779ce

update

14d35e9

Change-Id: I904f0120a728b4eb87226159f9c0cde3d18ef26a

RongtongJin changed the title ~~RIP-82 Implement Timer message, transaction message, and index based on RocksDB~~ [RIP-82] Implement Timer message, transaction message, and index based on RocksDB Oct 29, 2025

fuyou001 previously approved these changes Oct 29, 2025

View reviewed changes

fuyou001 self-requested a review October 29, 2025 05:19

drizzle.zk added 2 commits October 29, 2025 13:54

Merge remote-tracking branch 'github/develop' into opencore/develop_1022

fb58f56

update code

873732a

Change-Id: I21653d6bcad905daa8f10d12298d1d27535af9d9

zk-drizzle dismissed fuyou001’s stale review via 873732a October 29, 2025 06:57

update

9fa994b

Change-Id: I712b9cde9a18c730fd020ea76f05560e594a9edd

fuyou001 approved these changes Oct 29, 2025

View reviewed changes

RongtongJin reviewed Oct 30, 2025

View reviewed changes

majialoong reviewed Oct 30, 2025

View reviewed changes

	if (null == trs \|\| CollectionUtils.isEmpty(trs)) {
	if (CollectionUtils.isEmpty(trs)) {

[RIP-82] Implement Timer message, transaction message, and index based on RocksDB #9787

Are you sure you want to change the base?

[RIP-82] Implement Timer message, transaction message, and index based on RocksDB #9787

Conversation

zk-drizzle commented Oct 29, 2025

Brief Description

How Did You Test This Change?

Uh oh!

codecov-commenter commented Oct 29, 2025

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

majialoong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants