[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: 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 48.78%. Comparing base (1f387b2) to head (8f3d44e).
⚠️ Report is 1 commits behind head on develop.

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      49.81%   48.78%   -1.04%     
- Complexity     12731    12739       +8     
=============================================
  Files           1314     1324      +10     
  Lines          93924    96281    +2357     
  Branches       12047    12407     +360     
=============================================
+ Hits           46791    46972     +181     
- Misses         41437    43562    +2125     
- Partials        5696     5747      +51

☔ 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)) {

you are right ~

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?

Change-Id: Iafe10c007e9b5057212e2dcfb21671f970b1d344

majialoong · 2025-11-06T02:34:53Z

tools/src/main/java/org/apache/rocketmq/tools/command/broker/SwitchTimerEngineSubCommand.java

+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("c", "clusterName", true, "update which cluster");


Since either brokerAddr or clusterName must be specified, it would be more appropriate to use OptionGroup here and set Required to true. @zk-drizzle @fuyou001 @RongtongJin WDYT ?

I don't think it's necessary to require the cluster parameter, because when specifying a single broker explicitly, there's no need to set it.

@RongtongJin Thank you for your explanation. Yes, your description is correct; the code I marked might be misleading.

When specifying a broker, you don't need to set the clusterName.

However, you must choose between setting either the broker or the clusterName; this isn't mandatory in the current code.The current code doesn't enforce this; it only reminds the user to enter the e parameter.

Furthermore, other commands will use OptionGroup to remind the user that either the broker or the clusterName must be configured.

rocketmq/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicSubCommand.java

Lines 50 to 59 in 440341b

OptionGroup optionGroup = new OptionGroup();

Option opt = new Option("b", "brokerAddr", true, "create topic to which broker");

optionGroup.addOption(opt);

opt = new Option("c", "clusterName", true, "create topic to which cluster");

optionGroup.addOption(opt);

optionGroup.setRequired(true);

options.addOptionGroup(optionGroup);

If OptionGroup is used, the following message will be displayed:

It will remind the user that one of the broker and clusterName configurations must be set.

This may be more user-friendly and consistent with other commands.

@RongtongJin Please take a look when you have time. I think this is just an optimization and doesn't affect actual use. I would be happy to make adjustments in a future PR if possible. Thanks !

@RongtongJin Please take a look when you have time. I think this is just an optimization and doesn't affect actual use. I would be happy to make adjustments in a future PR if possible. Thanks !

Got it. This is indeed a better approach. Feel free to submit a PR for this change later. Thanks!

Change-Id: I40c1c7f54852ee033e0caa22573436ee505a27b8

Change-Id: Iae006b01608df4f363c63a0ae876dbf63cd56fbd

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

zk-drizzle and others added 2 commits October 28, 2025 17:27

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 previously 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

update

6f05121

Change-Id: Iafe10c007e9b5057212e2dcfb21671f970b1d344

zk-drizzle dismissed fuyou001’s stale review via 6f05121 November 5, 2025 08:34

fuyou001 previously approved these changes Nov 5, 2025

View reviewed changes

RongtongJin previously approved these changes Nov 6, 2025

View reviewed changes

majialoong reviewed Nov 6, 2025

View reviewed changes

ShannonDing previously approved these changes Nov 6, 2025

View reviewed changes

update

f1cc7ba

Change-Id: I40c1c7f54852ee033e0caa22573436ee505a27b8

zk-drizzle dismissed stale reviews from ShannonDing, RongtongJin, and fuyou001 via f1cc7ba November 6, 2025 11:47

zk-drizzle and others added 2 commits December 11, 2025 10:14

Merge branch 'develop' into opencore/develop_1022

f1fc2ad

optimize the code

8f3d44e

Change-Id: Iae006b01608df4f363c63a0ae876dbf63cd56fbd

RongtongJin approved these changes Dec 11, 2025

View reviewed changes

RongtongJin merged commit 7a08b4c into apache:develop Dec 11, 2025
11 checks passed

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

	OptionGroup optionGroup = new OptionGroup();

	Option opt = new Option("b", "brokerAddr", true, "create topic to which broker");
	optionGroup.addOption(opt);

	opt = new Option("c", "clusterName", true, "create topic to which cluster");
	optionGroup.addOption(opt);

	optionGroup.setRequired(true);
	options.addOptionGroup(optionGroup);

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

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

Uh oh!

Conversation

zk-drizzle commented Oct 29, 2025

Brief Description

How Did You Test This Change?

Uh oh!

codecov-commenter commented Oct 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

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!

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 Nov 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

codecov-commenter commented Oct 29, 2025 •

edited

Loading

majialoong Nov 6, 2025 •

edited

Loading