[ISSUE #4234]🚀Add RocketMQ Controller module; implement metadata management, RPC handling, and configuration management #4235
Conversation
…gement, RPC handling, and configuration management
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
rocketmq-rust-bot
added
AI review first
|
Warning Rate limit exceeded@mxsm has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 11 minutes and 35 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (29)
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull Request Overview
This PR introduces a new RocketMQ Controller module that replaces Java's DLedger with a Rust-based Raft consensus implementation. The controller provides high-availability cluster management including broker registration, topic configuration, and metadata synchronization.
Key changes:
- Complete Raft-based controller implementation using raft-rs library
- Multiple storage backends (RocksDB, file-based, in-memory) for metadata persistence
- Comprehensive metadata management for brokers, topics, configs, and replicas
- RPC server with codec for broker-controller communication
Reviewed Changes
Copilot reviewed 29 out of 30 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| rocketmq-controller/Cargo.toml | New package manifest with raft and storage dependencies |
| rocketmq-controller/src/lib.rs | Main library entry point with module exports |
| rocketmq-controller/src/config.rs | Controller configuration with Raft peer settings |
| rocketmq-controller/src/error.rs | Error types for controller operations |
| rocketmq-controller/src/manager.rs | Main controller manager coordinating all components |
| rocketmq-controller/src/raft/*.rs | Raft consensus layer implementation |
| rocketmq-controller/src/storage/*.rs | Storage backend abstractions and implementations |
| rocketmq-controller/src/metadata/*.rs | Metadata management for brokers, topics, configs, replicas |
| rocketmq-controller/src/processor/*.rs | Request processors for broker operations |
| rocketmq-controller/src/rpc/*.rs | RPC server and codec for network communication |
| Cargo.toml | Adds rocketmq-controller to workspace members |
| Cargo.lock | Dependency lock file updates |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| cargo build -p rocketmq-controller | ||
| ``` | ||
|
|
||
| ### 测试 |
There was a problem hiding this comment.
Chinese characters '测试' (meaning 'test') should be replaced with English 'Test' to match the rest of the documentation.
| ### 测试 | |
| ### Test |
| self.index.write().insert(key.to_string(), file_path); | ||
|
|
||
| // Save index periodically (every 10 operations) | ||
| if self.index.read().len() % 10 == 0 { |
There was a problem hiding this comment.
The modulo check len() % 10 == 0 has a subtle bug: it triggers on multiples of 10 (10, 20, 30), but also on 0 (empty index). This means it will attempt to save an empty index unnecessarily. Consider using len() > 0 && len() % 10 == 0 or moving this logic after the insert operation.
| if self.index.read().len() % 10 == 0 { | |
| if self.index.read().len() > 0 && self.index.read().len() % 10 == 0 { |
| // Get RocksDB property | ||
| let backend_info = db | ||
| .property_value("rocksdb.stats") | ||
| .unwrap_or(None) |
There was a problem hiding this comment.
The pattern .unwrap_or(None).unwrap_or_else(...) is redundant. Since property_value returns Result<Option<String>, Error>, you should use .ok().flatten().unwrap_or_else(|| ...) or simply handle the Result directly and then use unwrap_or_else on the Option.
| .unwrap_or(None) | |
| .ok() | |
| .flatten() |
| sync_state_set.last_update_timestamp = SystemTime::now() | ||
| .duration_since(SystemTime::UNIX_EPOCH) | ||
| .unwrap() | ||
| .as_secs(); | ||
| } | ||
|
|
||
| Ok(new_master) | ||
| } | ||
|
|
There was a problem hiding this comment.
Multiple calls to SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() are scattered throughout the file. This pattern should be extracted into a helper function to avoid code duplication and make error handling consistent. The .unwrap() could also panic if the system clock is set before Unix epoch.
| sync_state_set.last_update_timestamp = SystemTime::now() | |
| .duration_since(SystemTime::UNIX_EPOCH) | |
| .unwrap() | |
| .as_secs(); | |
| } | |
| Ok(new_master) | |
| } | |
| sync_state_set.last_update_timestamp = Self::current_unix_timestamp(); | |
| } | |
| Ok(new_master) | |
| } | |
| /// Returns the current UNIX timestamp in seconds, or 0 if system time is before UNIX_EPOCH. | |
| fn current_unix_timestamp() -> u64 { | |
| match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) { | |
| Ok(duration) => duration.as_secs(), | |
| Err(e) => { | |
| warn!("System time is before UNIX_EPOCH: {:?}", e); | |
| 0 | |
| } | |
| } | |
| } |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #4235 +/- ##
==========================================
+ Coverage 27.32% 27.67% +0.34%
==========================================
Files 586 609 +23
Lines 83771 86162 +2391
==========================================
+ Hits 22894 23845 +951
- Misses 60877 62317 +1440 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
…gement, RPC handling, and configuration management
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
…gement, RPC handling, and configuration management
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
rocketmq-rust-bot
added
approved
Which Issue(s) This PR Fixes(Closes)
Fixes #4234
Brief Description
How Did You Test This Change?
Summary by CodeRabbit