[ISSUE #658]🚀Support pull message consume #659
Conversation
|
Warning Rate limit exceeded@mxsm has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 40 minutes and 8 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. WalkthroughThe recent changes enhance the RocketMQ project by supporting pull message consumption and improving the messaging protocol. New structs and headers for pull message requests and responses were introduced, along with protocol updates. Configurations and handlers in the broker and processor layers were adjusted to accommodate these changes. Overall, these enhancements aim to streamline message consumption and improve broker-messaging functionalities. Changes
Sequence Diagram(s)Silent Pass Assessment against linked issues
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 as PR comments)
Additionally, you can add CodeRabbit Configration File (
|
|
🚀Thanks for your contribution🎉. CodeRabbit(AI) will review your code first🔥 |
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #659 +/- ##
==========================================
- Coverage 25.98% 25.57% -0.42%
==========================================
Files 234 236 +2
Lines 17079 17348 +269
==========================================
- Hits 4438 4436 -2
- Misses 12641 12912 +271 ☔ View full report in Codecov by Sentry. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Files selected for processing (14)
- rocketmq-broker/src/mqtrace/consume_message_hook.rs (1 hunks)
- rocketmq-broker/src/processor.rs (2 hunks)
- rocketmq-broker/src/processor/pull_message_processor.rs (1 hunks)
- rocketmq-broker/src/processor/pull_message_result_handler.rs (1 hunks)
- rocketmq-common/src/common/broker/broker_config.rs (2 hunks)
- rocketmq-remoting/src/protocol.rs (1 hunks)
- rocketmq-remoting/src/protocol/header.rs (1 hunks)
- rocketmq-remoting/src/protocol/header/message_operation_header/send_message_response_header.rs (1 hunks)
- rocketmq-remoting/src/protocol/header/namesrv/topic_operation_header.rs (1 hunks)
- rocketmq-remoting/src/protocol/header/pull_message_request_header.rs (1 hunks)
- rocketmq-remoting/src/protocol/header/pull_message_response_header.rs (1 hunks)
- rocketmq-remoting/src/protocol/remoting_command.rs (3 hunks)
- rocketmq-store/src/lib.rs (1 hunks)
- rocketmq/src/lib.rs (1 hunks)
Files skipped from review due to trivial changes (1)
- rocketmq-remoting/src/protocol/header.rs
Additional comments not posted (18)
rocketmq-store/src/lib.rs (1)
23-23: Thefiltermodule is now publicly accessible, which could be beneficial for extensibility. Ensure that all public interfaces are properly documented and consider restricting access to internal functionalities that should not be exposed.rocketmq/src/lib.rs (1)
23-42: Introduced a utility macrois_trait_implemented!to check trait implementations at compile-time. This is a robust addition for type-safe checks across the codebase.rocketmq-broker/src/mqtrace/consume_message_hook.rs (1)
22-22: Enhancing theConsumeMessageHooktrait withSync,Send, and'staticensures thread safety, crucial for the concurrent processing of messages in RocketMQ.rocketmq-broker/src/processor/pull_message_result_handler.rs (1)
25-38: Introduction ofPullMessageResultHandlertrait aligns with the PR's objectives to support pull message consumption. The methodhandleis well-designed to process various aspects of message handling, ensuring flexibility and robustness in message processing.rocketmq-broker/src/processor/pull_message_processor.rs (3)
32-36: The addition ofconsume_message_hook_vec,pull_message_result_handler, andbroker_configfields inPullMessageProcessorstruct ensures that the processor can handle hooks, results, and configurations effectively. This change supports the expansion of functionality for pull message processing.
45-53: Theprocess_requestmethod is now asynchronous and properly delegates toprocess_request_inner. This change enhances the function's ability to handle asynchronous operations and improves code readability by separating concerns.
55-91: Theprocess_request_innermethod includes several new checks and configurations, particularly for permissions and feature flags likelite_pull_message_enable. Ensure thorough testing for these conditional paths to verify the correct behavior under different configurations.rocketmq-remoting/src/protocol/header/pull_message_response_header.rs (2)
28-37: ThePullMessageResponseHeaderstruct is well-defined with optional fields that represent different aspects of message processing. This structure will facilitate detailed response management in pull message operations.
39-47: Constants defined for field names inPullMessageResponseHeaderenhance maintainability by avoiding hard-coded strings throughout the code. This is a good practice to prevent errors related to typos in string literals.rocketmq-remoting/src/protocol/header/namesrv/topic_operation_header.rs (1)
140-140: The addition of theDefaultderive attribute toTopicRequestHeaderis a sensible enhancement, facilitating easier instantiation of this struct with default values.rocketmq-common/src/common/broker/broker_config.rs (1)
145-145: Ensure proper default value handling forlite_pull_message_enable.The addition of the
lite_pull_message_enablefield with a default value oftrueis noted. This change aligns with the objective of supporting pull message consumption. Ensure that all configurations and conditional logic that depend on this new setting are updated accordingly.rocketmq-remoting/src/protocol/remoting_command.rs (2)
235-237: Added mutable setter foropaque.The addition of
set_opaque_mutprovides a mutable setter foropaque, enhancing flexibility in modifying this field within the same object context. This is a useful addition for scenarios whereopaquemight need to be updated after the initial creation of theRemotingCommandinstance.
368-380: Optimized method for decoding command headers.The new method
decode_command_custom_header_fastleverages trait bounds to ensure that the typeTimplementsFromMap,Default, andFastCodesHeader, which potentially optimizes the decoding process by using thedecode_fastmethod provided by theFastCodesHeadertrait. This is particularly useful for performance-critical paths where decoding happens frequently.rocketmq-remoting/src/protocol/header/pull_message_request_header.rs (3)
29-48: Introduction ofPullMessageRequestHeaderstruct.The
PullMessageRequestHeaderstruct is well-defined with fields that are essential for the pull message functionality such asconsumer_group,topic,queue_id, etc. The use ofOptionfor fields likesubscription,expression_type, andmax_msg_bytesprovides flexibility, allowing these fields to be optionally included based on the context of the request.
67-122: Implementation ofCommandCustomHeaderforPullMessageRequestHeader.The implementation of
CommandCustomHeaderforPullMessageRequestHeaderis comprehensive, covering all relevant fields and handling optional fields appropriately. This allows for a flexible and dynamic construction of command headers based on the actual data provided at runtime.
166-222: Fast encoding method forPullMessageRequestHeader.The
encode_fastmethod efficiently serializes thePullMessageRequestHeaderto aBytesMutbuffer. This method is crucial for performance optimization in scenarios where rapid serialization is required. The method handles optional fields gracefully, ensuring that only present values are written.Fast decoding method for
PullMessageRequestHeader.The
decode_fastmethod provides an efficient way to populate thePullMessageRequestHeaderfields from aHashMap. This method is essential for scenarios where headers need to be quickly reconstructed from their serialized forms. The method's implementation ensures robust error handling and defaults management, which is crucial for maintaining data integrity during the decoding process.rocketmq-broker/src/processor.rs (2)
74-74: Introduction ofpull_message_result_handlermodule.The addition of the
pull_message_result_handlermodule is a critical part of enhancing the broker's capability to handle pull message results effectively. This module is likely to contain logic that deals with the aggregation, filtering, and delivery of messages fetched through the pull mechanism.
146-150: Handling forPullMessageandLitePullMessagerequests.The update to handle
PullMessageandLitePullMessagerequests in theBrokerRequestProcessoris a significant enhancement aligning with the PR's objective to support pull message consumption. This ensures that pull requests are routed to the appropriate processor, facilitating efficient message fetching based on consumer demand.
| @@ -365,7 +365,7 @@ | |||
| } | |||
|
|
|||
| pub trait FastCodesHeader { | |||
| fn write_if_not_null(&mut self, out: &mut bytes::BytesMut, key: &str, value: &str) { | |||
| fn write_if_not_null(out: &mut bytes::BytesMut, key: &str, value: &str) { | |||
There was a problem hiding this comment.
Refactor the write_if_not_null method to improve clarity.
- fn write_if_not_null(out: &mut bytes::BytesMut, key: &str, value: &str) {
- if !value.is_empty() {
- RocketMQSerializable::write_str(out, true, key);
- RocketMQSerializable::write_str(out, false, key);
- }
+ fn write_if_not_null(out: &mut bytes::BytesMut, key: &str, value: &str) {
+ if !value.is_empty() {
+ RocketMQSerializable::write_str(out, true, key);
+ RocketMQSerializable::write_str(out, true, value);
+ }The current implementation seems to have a typo or logic error where the key is written twice instead of writing the key-value pair. This change corrects that mistake by ensuring that both the key and the value are correctly serialized into the output.
Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| fn write_if_not_null(out: &mut bytes::BytesMut, key: &str, value: &str) { | |
| fn write_if_not_null(out: &mut bytes::BytesMut, key: &str, value: &str) { | |
| if !value.is_empty() { | |
| RocketMQSerializable::write_str(out, true, key); | |
| RocketMQSerializable::write_str(out, true, value); | |
| } |
| impl Default for PullMessageProcessor { | ||
| fn default() -> Self { |
There was a problem hiding this comment.
The default implementation for PullMessageProcessor uses todo!(), which panics at runtime. This should be replaced with an actual implementation to prevent runtime crashes.
- todo!()
+ Self {
+ consume_message_hook_vec: Arc::new(Vec::new()),
+ pull_message_result_handler: Arc::new(Default::default()),
+ broker_config: Arc::new(Default::default()),
+ }Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| impl Default for PullMessageProcessor { | |
| fn default() -> Self { | |
| impl Default for PullMessageProcessor { | |
| fn default() -> Self { | |
| Self { | |
| consume_message_hook_vec: Arc::new(Vec::new()), | |
| pull_message_result_handler: Arc::new(Default::default()), | |
| broker_config: Arc::new(Default::default()), | |
| } |
| Self::write_if_not_null(out, "msgId", self.msg_id.to_string().as_str()); | ||
| Self::write_if_not_null(out, "queueId", self.queue_id.to_string().as_str()); | ||
| Self::write_if_not_null(out, "queueOffset", self.queue_offset.to_string().as_str()); | ||
| Self::write_if_not_null( | ||
| out, | ||
| "transactionId", | ||
| self.transaction_id.clone().as_deref().unwrap(), | ||
| ); | ||
| self.write_if_not_null( | ||
| Self::write_if_not_null( |
There was a problem hiding this comment.
The write_if_not_null method is used to serialize fields only if they are not null. However, calling unwrap() directly on Option types (transactionId, batchUniqId) can cause a panic if they are None. It's safer to handle these cases to avoid runtime errors.
- self.transaction_id.clone().as_deref().unwrap(),
- self.batch_uniq_id.clone().as_deref().unwrap(),
+ self.transaction_id.clone().as_deref().unwrap_or(""),
+ self.batch_uniq_id.clone().as_deref().unwrap_or(""),Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| Self::write_if_not_null(out, "msgId", self.msg_id.to_string().as_str()); | |
| Self::write_if_not_null(out, "queueId", self.queue_id.to_string().as_str()); | |
| Self::write_if_not_null(out, "queueOffset", self.queue_offset.to_string().as_str()); | |
| Self::write_if_not_null( | |
| out, | |
| "transactionId", | |
| self.transaction_id.clone().as_deref().unwrap(), | |
| ); | |
| self.write_if_not_null( | |
| Self::write_if_not_null( | |
| Self::write_if_not_null(out, "msgId", self.msg_id.to_string().as_str()); | |
| Self::write_if_not_null(out, "queueId", self.queue_id.to_string().as_str()); | |
| Self::write_if_not_null(out, "queueOffset", self.queue_offset.to_string().as_str()); | |
| Self::write_if_not_null( | |
| out, | |
| "transactionId", | |
| self.transaction_id.clone().as_deref().unwrap_or(""), | |
| ); | |
| Self::write_if_not_null( | |
| out, | |
| "batchUniqId", | |
| self.batch_uniq_id.clone().as_deref().unwrap_or(""), | |
| ); | |
| Self::write_if_not_null( |
…ader.rs Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
|
🚀Thanks for your contribution🎉. CodeRabbit(AI) will review your code first🔥 |
There was a problem hiding this comment.
Actionable comments posted: 0
Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Files selected for processing (1)
- rocketmq-remoting/src/protocol/header/pull_message_response_header.rs (1 hunks)
Files skipped from review as they are similar to previous changes (1)
- rocketmq-remoting/src/protocol/header/pull_message_response_header.rs
|
🚀Thanks for your contribution🎉. CodeRabbit(AI) will review your code first🔥 |
Which Issue(s) This PR Fixes(Closes)
Fixes #658
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
New Features
PullMessageResponseHeaderfor handling message pulling with fast serialization capabilities.lite_pull_message_enableoption.Improvements
ConsumeMessageHookwith additional trait constraints for better concurrency support.PullMessageProcessorwith new fields and methods to improve message processing.Refactor
SendMessageResponseHeaderto use static method calls for improved code clarity.filtermodule to be publicly accessible.Documentation