[ISSUE #1939]🔖PopMessageProcessor supports process_request handle-4🚀 #1950
Conversation
WalkthroughThis pull request introduces enhancements to the RocketMQ Rust broker's message processing infrastructure, focusing on the Changes
Assessment against linked issues
Possibly related PRs
Suggested labels
Suggested reviewers
Poem
📜 Recent review detailsConfiguration used: .coderabbit.yaml 📒 Files selected for processing (5)
🚧 Files skipped from review as they are similar to previous changes (5)
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
|
|
🔊@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: 3
🧹 Nitpick comments (11)
rocketmq-common/src/common/broker/broker_config.rs (4)
58-104: Multiple factory methods might be consolidated or DRYed.
new,new_with_container,new_with_params, andnew_with_container_paramsshare common logic. Consider extracting shared code to reduce duplication if usage patterns allow.
106-189: BrokerConfig: Large struct with numerous fields.
- This struct organizes many fields. Depending on how these are used, consider grouping related settings into sub-structs to improve readability.
- The new field
pop_response_return_actual_retry_topicis a clear addition. Ensure it’s consistently read/used by the relevant pop logic to avoid confusion.
285-575:get_propertiesmethod: Potential performance overhead.
- This method collects fields into a
HashMapeach time. If called frequently, consider caching or using a lazy approach.- The property keys are user-facing. Confirm that naming is consistent with any existing config standards.
578-583:default_broker_nameis straightforward.Reading
LOCAL_HOST_NAMEis fine. Consider logging or warning if the hostname is not found and you use "DEFAULT_BROKER" as fallback.rocketmq-broker/src/processor/pop_message_processor.rs (3)
84-85: Arc references to the offset managers.Line 84-85 indicates new fields in the
PopMessageProcessor. Combining them under a single struct could simplify passing them around if usage grows.
Line range hint
447-566: Pop from retry topic logic duplication.These blocks for v1 vs. v2 retry topic might be made DRY by extracting a helper function. The logic is almost identical except for the topic name.
660-681:pop_msg_from_topic: Round-robin logic.You’re doing
(random_q + index as i32) % topic_config.read_queue_nums. This is reasonable for distributing loads. Consider if the distribution is truly random or if it needs more uniform randomization.rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs (2)
17-18: Unused variables lint override.The
#![allow(unused_variables)]annotation may hide accidental oversights. If feasible, consider removing it and addressing any unused variables directly.
268-281: Unimplemented placeholder update method.The
updatemethod signature suggests important future logic. For better clarity and tracking, either implement its functionality or add a descriptive TODO comment within the method body.rocketmq-broker/src/processor/pull_message_processor.rs (1)
683-691: Use a more descriptive or context-aware name formessage_filter.
The variable namemessage_filteris accurate yet might be mistaken for local filtering only. Consider clarifying the scope or purpose, for instanceretry_or_expression_filter, to better reflect that it can be one of two filter types.rocketmq-store/src/message_store/default_message_store.rs (1)
965-972: Gracefully handle filtering errors inis_matched_by_commit_log.
While the code correctly continues when a message fails the commit-log check, consider logging or aggregating counts of rejections for deeper insight into how many messages are skipped due to filtering.
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (13)
rocketmq-broker/src/broker_runtime.rs(1 hunks)rocketmq-broker/src/long_polling/long_polling_service/pop_long_polling_service.rs(1 hunks)rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs(1 hunks)rocketmq-broker/src/processor/default_pull_message_result_handler.rs(2 hunks)rocketmq-broker/src/processor/pop_message_processor.rs(18 hunks)rocketmq-broker/src/processor/processor_service/pop_buffer_merge_service.rs(1 hunks)rocketmq-broker/src/processor/pull_message_processor.rs(2 hunks)rocketmq-broker/src/processor/pull_message_result_handler.rs(2 hunks)rocketmq-common/src/common/broker/broker_config.rs(1 hunks)rocketmq-remoting/src/protocol/header/extra_info_util.rs(1 hunks)rocketmq-store/src/base/get_message_result.rs(1 hunks)rocketmq-store/src/log_file.rs(1 hunks)rocketmq-store/src/message_store/default_message_store.rs(3 hunks)
✅ Files skipped from review due to trivial changes (1)
- rocketmq-remoting/src/protocol/header/extra_info_util.rs
🔇 Additional comments (27)
rocketmq-common/src/common/broker/broker_config.rs (5)
18-20: Imports look appropriate.
You are using Any in this file, but it does not appear to be referenced later in the code. Verify if you plan to use dynamic type checks in the future.
35-46: Consider verifying the lazy-static usage for concurrency safety and overhead.
lazy_static! macros are suitable for one-time initialization, but if these values ever need dynamic updates, consider other approaches. If static references remain constant, this usage is fine.
48-56: Struct BrokerIdentity: Good encapsulation.
Struct fields are clearly named. Watch for future feature creep—some booleans may evolve into enumerations or consolidated fields.
191-283: Default trait implementation is comprehensive.
- Good job providing sensible defaults for each field.
- Consider referencing environment variables or a configuration file for broker IP/port to reduce potential coupling with system environment calls.
584-603: TopicQueueConfig and TimerWheelConfig usage.
They look fine. For extremely large-scale usage, double-check that defaults (e.g., timer_wheel_enable: bool) work for typical production scenarios.
rocketmq-broker/src/processor/pop_message_processor.rs (12)
37-37: Ensure message decoding usage is consistent.
Line 37 references rocketmq_common::common::message::message_decoder, but usage occurs further down. Confirm it’s actually utilized in the branch that’s not yet implemented.
45-45: ExtraInfoUtil usage.
Ensure consistent usage of ExtraInfoUtil across all relevant code paths that handle extended fields for pop messages.
73-75: New consumer offset and order managers.
The additional managers (ConsumerOffsetManager, ConsumerOrderInfoManager) are introduced. Verify correct initialization in other places (e.g., broker_runtime).
95-95: pop_inflight_message_counter usage.
This is a powerful concurrency tracking tool. Ensure atomic increments/decrements match message lifecycle to avoid memory leaks.
101-102: Constructor parameter addition.
consumer_offset_manager and consumer_order_info_manager are now mandatory. Verify any existing tests or code paths that call PopMessageProcessor::new(...).
112-114: Initialization block uses ArcMut::new(...).
Double-check concurrency patterns, as ArcMut is a specialized type. Confirm that no data race occurs if PopLongPollingService or PopBufferMergeService is mutated concurrently.
421-423: Creating GetMessageResult with new_result_size() is appropriate.
Ensure the max_msg_nums argument is validated to avoid large memory allocations.
436-436: Mapping message_filter to Arc.
Converting Box<dyn MessageFilter> to Arc<Box<dyn MessageFilter>>> is typical for concurrency. Ensure all implementors are thread-safe.
653-654: Additional optional message_filter.
Re-check usage for topics that might require different filters or no filter at all.
684-719: pop_msg_from_topic_by_name: Good fallback.
Simplifies code by reusing pop_msg_from_topic. Ensure no infinite loops if the same fallback is called repeatedly.
953-961: is_pop_should_stop unimplemented.
Recommend a sane default (e.g., returning true or false) plus logging.
962-991: get_pop_offset logic.
Possible race conditions if pop_buffer_merge_service.get_latest_offset is updated concurrently. Confirm consistency with consumer_offset_manager.
rocketmq-broker/src/long_polling/long_polling_service/pop_long_polling_service.rs (1)
53-53: Changed message_filter type from Arc<Option<Box<dyn MessageFilter>>> to Option<Arc<Box<dyn MessageFilter>>>>.
This is an improvement—logic for optional filters is typically easier to handle than an Arc<Option<...>>. Check all call sites to ensure they handle None properly now.
rocketmq-broker/src/processor/pull_message_result_handler.rs (1)
71-71: Shift to Arc<Box<dyn MessageFilter>>.
Supports concurrency more consistently. Ensure implementing objects are sync-safe to avoid data races.
rocketmq-store/src/base/get_message_result.rs (1)
231-233: Expose message_queue_offset with caution.
Returning a reference to an internal vector is fine in this context, but be mindful if any future modifications might require more controlled access, such as returning a slice or an iterator for safety.
rocketmq-store/src/log_file.rs (1)
249-249: Ensure correct usage of Arc<Box> throughout.
Updating the filter parameter type from Option<&dyn MessageFilter> to Option<Arc<Box<dyn MessageFilter>>> is consistent with the rest of the codebase, allowing shared ownership. Verify all call sites to confirm they properly supply this new type.
rocketmq-broker/src/processor/default_pull_message_result_handler.rs (2)
122-122: Adopt Arc<Box> for concurrency.
The signature update aligns with concurrent usage patterns. Ensure that the filter is safely shared among threads without risking data races.
237-237: Suspend logic uses the updated filter reference.
Verify that suspending the pull request with message_filter maintains the expected lifetime and concurrency guarantees, especially under high load or frequently suspended requests.
rocketmq-broker/src/processor/pull_message_processor.rs (1)
744-744: Confirm message_filter.clone() usage across the codebase.
Passing message_filter.clone() ensures that each call site receives a shared reference, but verify you’ve replaced all old references to match the new Arc-wrapped type to avoid any potential mismatches.
rocketmq-broker/src/broker_runtime.rs (1)
566-574: Ensure consistency when adding new parameters to PopMessageProcessor.
These newly introduced fields (consumer_offset_manager, consumer_order_info_manager, pop_inflight_message_counter) expand the processor’s capabilities significantly. Confirm the initialization of these fields is always in sync with their usage in PopMessageProcessor logic (e.g., verifying none remain unused or incorrectly cloned).
rocketmq-store/src/message_store/default_message_store.rs (2)
797-797: Check for None usage where a filter might be required.
The updated parameter signature now expects an Option<Arc<Box<dyn MessageFilter>>>. Ensure that any call sites that intentionally pass None remain valid if filtering is mandatory at any point in the call flow.
Line range hint 934-938: Validate is_matched_by_consume_queue usage before commit log checks.
This block correctly short-circuits early if the consume-queue-level filter fails, saving resources. Just confirm that skipping the commit-log check is desired and consistent with the overall filtering logic.
| topic, | ||
| queue_id, | ||
| offset, | ||
| request_header.max_msg_nums as i32 | ||
| - get_message_result.message_mapped_list().len() as i32, | ||
| 1024 * 1024, | ||
| message_filter, | ||
| ) | ||
| .await; | ||
| match get_message_result_in { | ||
| None => { | ||
| let num = | ||
| self.message_store.get_max_offset_in_queue(topic, queue_id) | ||
| - atomic_offset.load(Ordering::Acquire) | ||
| + atomic_rest_num.load(Ordering::Acquire); | ||
| return num; | ||
| } | ||
| Some(value) => value, | ||
| } | ||
| } | ||
| _ => result, | ||
| } | ||
| }; | ||
| if !result_inner.message_mapped_list().is_empty() { | ||
| if is_order { | ||
| self.consumer_order_info_manager.update( | ||
| request_header.attempt_id.clone().unwrap_or_default(), | ||
| is_retry, | ||
| topic, | ||
| &request_header.consumer_group, | ||
| queue_id, | ||
| pop_time, | ||
| request_header.invisible_time, | ||
| result_inner.message_queue_offset().clone(), | ||
| order_count_info, | ||
| ); | ||
| self.consumer_offset_manager.commit_offset( | ||
| channel.remote_address(), | ||
| &request_header.consumer_group, | ||
| topic, | ||
| queue_id, | ||
| final_offset, | ||
| ); | ||
| } else if !self.append_check_point( | ||
| request_header, | ||
| topic, | ||
| revive_qid, | ||
| queue_id, | ||
| final_offset, | ||
| &result_inner, | ||
| pop_time as i64, | ||
| self.broker_config.broker_name.as_str(), | ||
| ) { | ||
| return atomic_rest_num.load(Ordering::Acquire) | ||
| + result_inner.message_count() as i64; | ||
| } | ||
|
|
||
| ExtraInfoUtil::build_start_offset_info( | ||
| start_offset_info, | ||
| topic, | ||
| queue_id, | ||
| final_offset, | ||
| ); | ||
| ExtraInfoUtil::build_msg_offset_info( | ||
| msg_offset_info, | ||
| topic, | ||
| queue_id, | ||
| result_inner.message_queue_offset().clone(), | ||
| ); | ||
| } else { | ||
| unimplemented!() | ||
| } | ||
|
|
||
| atomic_rest_num.fetch_add( | ||
| result_inner.max_offset() - result_inner.next_begin_offset(), | ||
| Ordering::AcqRel, | ||
| ); | ||
| let broker_name = self.broker_config.broker_name.as_str(); | ||
| for msg in result_inner.message_mapped_list() { | ||
| if self.broker_config.pop_response_return_actual_retry_topic || !is_retry { | ||
| //get_message_result.message_mapped_list().push(msg.clone()); | ||
| } else { | ||
| let mut bytes = msg.get_bytes().unwrap_or_default(); | ||
| let message_ext_list = | ||
| message_decoder::decodes_batch(&mut bytes, true, false); | ||
| unimplemented!("PopMessageProcessor pop_msg_from_queue") | ||
| } | ||
| } | ||
| self.pop_inflight_message_counter | ||
| .increment_in_flight_message_num( | ||
| topic, | ||
| &request_header.consumer_group, | ||
| queue_id, | ||
| result_inner.message_count() as i64, | ||
| ); | ||
| } | ||
| } | ||
| self.queue_lock_manager().unlock_with_key(lock_key).await; | ||
| atomic_rest_num.load(Ordering::Acquire) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
pop_msg_from_queue: Complex logic mixing offset, concurrency, and queue locking.
- Hard-coded calls to
unimplemented!()break code flow if triggered. Provide at least a graceful error or handle the scenario if the function is essential. - The queue lock approach is correct for concurrency but consider any performance overhead with
try_lock_with_key. - Ensure consistent offset commits if an error occurs midway.
| fn append_check_point( | ||
| &self, | ||
| request_header: &PopMessageRequestHeader, | ||
| topic: &str, | ||
| revive_qid: i32, | ||
| queue_id: i32, | ||
| offset: i64, | ||
| get_message_tmp_result: &GetMessageResult, | ||
| pop_time: i64, | ||
| broker_name: &str, | ||
| ) -> bool { | ||
| unimplemented!() | ||
| } | ||
|
|
There was a problem hiding this comment.
append_check_point unimplemented.
Calls to an unimplemented method can hinder functionality. Return an error or default behavior instead, to prevent runtime panics.
- unimplemented!()
+ // Possibly return a boolean or store an event log:
+ trace!("append_check_point is not implemented yet");
+ false📝 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 append_check_point( | |
| &self, | |
| request_header: &PopMessageRequestHeader, | |
| topic: &str, | |
| revive_qid: i32, | |
| queue_id: i32, | |
| offset: i64, | |
| get_message_tmp_result: &GetMessageResult, | |
| pop_time: i64, | |
| broker_name: &str, | |
| ) -> bool { | |
| unimplemented!() | |
| } | |
| fn append_check_point( | |
| &self, | |
| request_header: &PopMessageRequestHeader, | |
| topic: &str, | |
| revive_qid: i32, | |
| queue_id: i32, | |
| offset: i64, | |
| get_message_tmp_result: &GetMessageResult, | |
| pop_time: i64, | |
| broker_name: &str, | |
| ) -> bool { | |
| trace!("append_check_point is not implemented yet"); | |
| false | |
| } |
| pub fn get_latest_offset(&self, _lock_key: &str) -> i64 { | ||
| unimplemented!("Not implemented yet"); | ||
| } |
There was a problem hiding this comment.
get_latest_offset is unimplemented.
Provide at least a stub or default logic to avoid panics in production. Leaving this unimplemented can break the code path in PopMessageProcessor.
- unimplemented!("Not implemented yet");
+ // Return a default offset or retrieve real offset
+ 0📝 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.
| pub fn get_latest_offset(&self, _lock_key: &str) -> i64 { | |
| unimplemented!("Not implemented yet"); | |
| } | |
| pub fn get_latest_offset(&self, _lock_key: &str) -> i64 { | |
| // Return a default offset or retrieve real offset | |
| 0 | |
| } |
|
🔊@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 #1939
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
New Features
Bug Fixes
Documentation
Refactor
Chores