Repository navigation
[ISSUE #3428]♻️Refactor DefaultTransactionalMessageCheckListener to simplify structure and enhance message handling - #3429
Conversation
…implify structure and enhance message handling
|
""" WalkthroughThe code refactors the Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant DefaultListener
participant BrokerClient
Client->>DefaultListener: new()
DefaultListener->>BrokerClient: store ArcMut<Broker2Client>
Client->>DefaultListener: send_check_message(msg_ext)
DefaultListener->>BrokerClient: send check message (async)
BrokerClient-->>DefaultListener: RocketMQResult
Client->>DefaultListener: resolve_half_msg(msg_ext)
DefaultListener->>BrokerClient: resolve half message (async)
BrokerClient-->>DefaultListener: RocketMQResult
Assessment against linked issues
Assessment against linked issues: Out-of-scope changesNo out-of-scope changes were found. Poem
📜 Recent review detailsConfiguration used: .coderabbit.yaml 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
⏰ Context from checks skipped due to timeout of 90000ms (9)
✨ Finishing Touches
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. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. 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
There was a problem hiding this comment.
Pull Request Overview
This pull request refactors the transactional message check listener structure to simplify its design and enhance message handling for transactional messages.
- Added async methods (send_check_message and resolve_half_msg) to the TransactionalMessageCheckListenerInner trait for improved message checking.
- Modified DefaultTransactionalMessageCheckListener to remove the inner field in favor of directly using broker_client and reorganized the inner struct’s definition.
Reviewed Changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| rocketmq-broker/src/transaction/transactional_message_check_listener.rs | Added documentation and new async method signatures to the trait to standardize message checking and resolution. |
| rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs | Refactored the DefaultTransactionalMessageCheckListener by removing the inner field, updating the constructor, and converting resolve_half_msg to an async function. |
| } | ||
|
|
||
| pub fn resolve_half_msg(&self, _msg_ext: MessageExt) -> rocketmq_error::RocketMQResult<()> { | ||
| async fn resolve_half_msg(&self, _msg_ext: MessageExt) -> rocketmq_error::RocketMQResult<()> { |
There was a problem hiding this comment.
The use of unimplemented!("resolve_half_msg") may lead to runtime panics if this async function is invoked. Consider providing an implementation or clearly documenting that this code path should not be reached in production.
| /* broker_config: Arc<BrokerConfig>, | ||
| producer_manager: Arc<ProducerManager>,*/ | ||
| broker_client: Broker2Client, | ||
| broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>, | ||
| ) -> Self { | ||
| Self { | ||
| /*broker_config, | ||
| producer_manager,*/ |
There was a problem hiding this comment.
[nitpick] There are several commented-out lines in the TransactionalMessageCheckListenerInner::new method. Removing these commented code blocks will improve readability and maintainability.
| /* broker_config: Arc<BrokerConfig>, | |
| producer_manager: Arc<ProducerManager>,*/ | |
| broker_client: Broker2Client, | |
| broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>, | |
| ) -> Self { | |
| Self { | |
| /*broker_config, | |
| producer_manager,*/ | |
| broker_client: Broker2Client, | |
| broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>, | |
| ) -> Self { | |
| Self { |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🔭 Outside diff range comments (3)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs (3)
172-180:⚠️ Potential issueHandle the error from
check_producer_transaction_state.The error returned by
check_producer_transaction_stateis currently ignored. This could hide important failures in transaction state checking.- let _ = self + if let Err(e) = self .broker_client .check_producer_transaction_state( group_id.as_ref().unwrap(), &mut channel, header, msg_ext, ) - .await; + .await { + warn!("Failed to check producer transaction state for groupId={:?}: {:?}", group_id, e); + return Err(e); + }
175-175:⚠️ Potential issueAvoid potential panic from
unwrap().While
group_idis used to get the channel, callingunwrap()ongroup_id.as_ref()could panic ifgroup_idisNone. The channel existence doesn't guaranteegroup_idisSome.- group_id.as_ref().unwrap(), + group_id.as_ref().ok_or_else(|| { + rocketmq_error::Error::Custom("Producer group ID is missing".to_string()) + })?,
190-199:⚠️ Potential issueImplement the
resolve_half_msgmethod.The method is currently unimplemented and will panic at runtime. Based on the commented code, it seems the intended behavior is to spawn an async task to send check messages.
Would you like me to help implement this method based on the commented logic, or open an issue to track this TODO?
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs(5 hunks)rocketmq-broker/src/transaction/transactional_message_check_listener.rs(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
rocketmq-broker/src/transaction/transactional_message_check_listener.rs (1)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs (2)
send_check_message(127-188)resolve_half_msg(190-199)
⏰ Context from checks skipped due to timeout of 90000ms (9)
- GitHub Check: build (windows-latest, nightly)
- GitHub Check: build (macos-latest, nightly)
- GitHub Check: build (windows-latest, stable)
- GitHub Check: build (macos-latest, stable)
- GitHub Check: build (ubuntu-latest, stable)
- GitHub Check: build (ubuntu-latest, nightly)
- GitHub Check: test
- GitHub Check: build
- GitHub Check: auto-approve
🔇 Additional comments (2)
rocketmq-broker/src/transaction/transactional_message_check_listener.rs (1)
44-79: Well-structured trait additions!The new async methods
send_check_messageandresolve_half_msgare properly designed with clear documentation, appropriate error handling, and consistent signatures. The async nature is suitable for these I/O-bound operations.rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs (1)
43-76: LGTM! Simplified struct design.The refactoring to remove the inner field and directly embed
broker_clientsimplifies the structure and reduces unnecessary indirection. The Clone implementation and constructor are correctly updated.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #3429 +/- ##
=======================================
Coverage 25.84% 25.84%
=======================================
Files 542 542
Lines 77056 77032 -24
=======================================
Hits 19912 19912
+ Misses 57144 57120 -24 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
…ssage_check_listener.rs Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
|
🔊@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💥. |
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs (1)
190-199:⚠️ Potential issueAddress the unimplemented! macro to prevent runtime panics.
The
unimplemented!("resolve_half_msg")will cause runtime panics if this method is called. Either provide a proper implementation or usetodo!()with clear documentation about when this will be implemented.Consider this implementation approach:
- async fn resolve_half_msg(&self, _msg_ext: MessageExt) -> rocketmq_error::RocketMQResult<()> { - /*let this = self.clone(); - tokio::spawn(async move { - if let Err(e) = this.send_check_message(msg_ext).await { - warn!("Send check m_msg_extfailed: {:?}", e); - } - }); - Ok(())*/ - unimplemented!("resolve_half_msg") - } + async fn resolve_half_msg(&self, msg_ext: MessageExt) -> rocketmq_error::RocketMQResult<()> { + let this = self.clone(); + tokio::spawn(async move { + if let Err(e) = this.send_check_message(msg_ext).await { + warn!("Send check message failed: {:?}", e); + } + }); + Ok(()) + }
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs(5 hunks)
🧰 Additional context used
🪛 GitHub Actions: CI
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs
[error] 207-207: cargo fmt check failed. Code formatting issues detected. Run 'cargo fmt' to fix formatting.
⏰ Context from checks skipped due to timeout of 90000ms (4)
- GitHub Check: automerge
- GitHub Check: test
- GitHub Check: build
- GitHub Check: auto-approve
🔇 Additional comments (3)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs (3)
47-47: LGTM: Direct broker_client embedding simplifies structure.The refactoring to directly embed
broker_clientasArcMut<Broker2Client>instead of using an inner struct is a good simplification that reduces indirection.
53-53: LGTM: Constructor and clone implementation properly updated.The updates to both the
Cloneimplementation and constructor correctly handle the directbroker_clientfield with properArcMutwrapping.Also applies to: 74-74
127-188: LGTM: Well-implemented async message checking logic.The
send_check_messagemethod is well-structured with proper error handling, message property extraction, and asynchronous broker client communication. The logic correctly handles missing channels with appropriate warnings.
| fn as_any(&self) -> &dyn Any { | ||
| self | ||
| } | ||
|
|
||
| fn as_any_mut(&mut self) -> &mut dyn Any { | ||
| self | ||
| } | ||
| } |
There was a problem hiding this comment.
Fix formatting issues flagged by CI.
The pipeline indicates formatting issues at line 207. Run cargo fmt to resolve the formatting problems.
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}Committable suggestion skipped: line range outside the PR's diff.
🧰 Tools
🪛 GitHub Actions: CI
[error] 207-207: cargo fmt check failed. Code formatting issues detected. Run 'cargo fmt' to fix formatting.
🤖 Prompt for AI Agents
In
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs
around lines 201 to 208, the code formatting does not comply with the project's
style guidelines as flagged by CI. Run `cargo fmt` on this file to automatically
fix the formatting issues, ensuring consistent indentation, spacing, and line
breaks according to Rust's standard formatting rules.
|
🔊@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 #3428
Brief Description
How Did You Test This Change?
Summary by CodeRabbit