[ISSUE #1881]🚀Implement AckMessageProcessor#ackOrderly method logic #1882
Conversation
WalkthroughThis pull request introduces enhancements to the RocketMQ broker's message processing and consumer management components. The changes primarily focus on the Changes
Assessment against linked issues
Possibly related PRs
Suggested labels
Suggested reviewers
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
Documentation and Community
|
rocketmq-rust-bot
requested review from
rocketmq-rust-bot,
SpaceXCN and
TeslaRustor
RocketmqRustBot
added
Difficulty level/Moderate
|
🔊@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
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (6)
rocketmq-broker/src/processor/ack_message_processor.rs (2)
86-88: Constructor signature expanded
You're now passing new Arc-based parameters into AckMessageProcessor. Verify that initialization remains synchronous and side-effect free to avoid potential deadlocks on startup.
340-341: Batch offset loop branch
Any offset belonging to POP_ORDER_REVIVE_QUEUE branches again to ack_orderly. Consider logging or clarifying the reason for the queue assignment to reduce debugging overhead.rocketmq-broker/src/processor/pop_message_processor.rs (1)
54-65: New notify_message_arriving method
The newly added method logs a message arrival event. Currently unimplemented beyond logging. Make sure to eventually implement the relevant logic or remove the method if only logging is needed.Would you like me to propose a minimal implementation to handle message arrival notifications?
rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs (3)
17-17: Unused variables allowed
The #![allow(unused_variables)] attribute is a temporary workaround for in-progress logic. Clean up unused references once features are finalized.
245-255: commit_and_next method stub
This function is crucial for ordering logic, yet unimplemented. Consider at least returning an “illegal offset” or a default for easier debugging.
257-266: check_block method stub
Same as above. This method is also unimplemented. If it’s a placeholder, confirm such placeholders are tracked in your backlog. Otherwise, returning a default might reduce confusion.
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
rocketmq-broker/src/broker_runtime.rs(1 hunks)rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs(2 hunks)rocketmq-broker/src/processor/ack_message_processor.rs(9 hunks)rocketmq-broker/src/processor/pop_message_processor.rs(1 hunks)
🔇 Additional comments (11)
rocketmq-broker/src/processor/ack_message_processor.rs (10)
70-72: Fields integrated for offset & consumer order management
The newly introduced fields (consumer_offset_manager, consumer_order_info_manager, and pop_message_processor) in the AckMessageProcessor facilitate offset tracking and acknowledgment logic for both normal and orderly consumption. This design choice nicely centralizes related managers.
98-100: Storing fields in the struct
Storing the newly passed managers properly. No issues here. Just ensure you handle concurrency carefully if you later expand on any parallel initialization tasks.
275-276: Asynchronous call to ack_orderly
The invocation of ack_orderly for an orderly queue is triggered here. Ensure that the resolved future is handled properly, especially in case of errors, so you don't leave partial states behind.
Line range hint 446-456: ack_orderly method signature
Switching ack_orderly to async looks good. Ensure to cover potential concurrency conflicts with the pop_message_processor if the method is called frequently.
[approve]
457-464: Constructing lock key
Lock key generation is straightforward and consistent with typical patterns for queue locks. No issues spotted; local key usage is clear.
465-471: Offset mismatch check
We skip acknowledgment if old_offset > ack_offset. This ensures no stale or out-of-order offsets. The logic is correct and prevents rollback-like scenarios.
477-482: Re-check old_offset
Double-checking offset after lock acquisition is prudent and prevents race conditions. Good concurrency practice!
490-538: Result match for next_offset
The match handles offset state transitions well. The ordering::Less, Equal, and Greater branches are well structured. Ensure that partial error states (like large negative offsets) are addressed or at least logged.
539-544: Unlock key & decrement inflight
You free the lock after successful offset handling and reduce the inflight count. This pattern is correct for avoiding partial states.
Line range hint 572-574: Registering AckMessageProcessor in init_processor
These lines show the injection of consumerOffsetManager, popMessageProcessor, and consumerOrderInfoManager. The usage is consistent with the AckMessageProcessor constructor changes. Looks good!
rocketmq-broker/src/broker_runtime.rs (1)
572-574: AckMessageProcessor creation
Here, you pass additional references that broaden the AckMessageProcessor’s responsibilities in offset and consumer order management. Confirm that the references remain valid throughout the broker’s lifecycle.
| let next_offset = self.consumer_order_info_manager.commit_and_next( | ||
| &consume_group, | ||
| &topic, | ||
| q_id, | ||
| ack_offset as u64, | ||
| pop_time as u64, | ||
| ); |
There was a problem hiding this comment.
commit_and_next
The unwrapped function call to consumer_order_info_manager.commit_and_next is presumably fundamental to your ordering logic. Ensure robust error-handling, especially for edge cases like negative offsets or missing topic.
| .pop_message_processor | ||
| .queue_lock_manager() | ||
| .try_lock_with_key(lock_key.clone()) | ||
| .await | ||
| {} |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Attempt to acquire queue lock
The loop while !try_lock_with_key(...) can lead to busy-wait. Consider a small async delay or re-check to avoid hogging the runtime if locks are highly contested.
- while !self
- .pop_message_processor
- .queue_lock_manager()
- .try_lock_with_key(lock_key.clone())
- .await
- {}
+ loop {
+ if self
+ .pop_message_processor
+ .queue_lock_manager()
+ .try_lock_with_key(lock_key.clone())
+ .await {
+ break;
+ }
+ tokio::time::sleep(Duration::from_millis(5)).await;
+ }Committable suggestion skipped: line range outside the PR's diff.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #1882 +/- ##
==========================================
- Coverage 28.48% 28.44% -0.05%
==========================================
Files 475 475
Lines 66809 66929 +120
==========================================
+ Hits 19032 19035 +3
- Misses 47777 47894 +117 ☔ View full report in Codecov by Sentry. |
rocketmq-rust-bot
added
approved
Which Issue(s) This PR Fixes(Closes)
Fixes #1881
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
New Features
AckMessageProcessor.ConsumerOrderInfoManager.PopMessageProcessor.Bug Fixes
Documentation