[ISSUE #972]🔥Supports client clusting consume🚀

[mxsm]

Which Issue(s) This PR Fixes(Closes)

Fixes #972

Brief Description

How Did You Test This Change?

Summary by CodeRabbit

• New Features

  • Enhanced message consumption flexibility by allowing consumers to receive messages with any tag.
  • Introduced new methods for querying consumer offsets and pulling messages, improving message handling capabilities.
  • Added a new struct MessageClientExt with comprehensive methods for managing message properties.
  • Implemented asynchronous message processing and improved error handling in pull callbacks.

• Bug Fixes

  • Adjusted method signatures to improve memory management and concurrency handling.

• Documentation

  • Updated method signatures and added comments for clarity and better usability.

• Refactor

  • Simplified handling of MessageQueue parameters across multiple methods.
  • Enhanced the ConsumeMessageContext and FilterMessageContext structures for better memory management.

[coderabbitai]

Caution

Review failed

The pull request is closed.

Walkthrough

The changes in this pull request encompass significant modifications to the RocketMQ client implementation, focusing on enhancing message consumption, improving memory management, and refining the handling of offsets and message properties. Key updates include adjustments to method signatures, the introduction of new structs and fields, and the reorganization of module visibility. These changes aim to facilitate more efficient and flexible message processing within the client.

Changes

Files	Change Summary
rocketmq-client/examples/quickstart/consumer.rs	Modified message filtering tag and updated method signatures for message consumption handling.
rocketmq-client/src/base/client_config.rs	Changed parameter and return types in queue_with_namespace method.
rocketmq-client/src/consumer.rs	Changed visibility of several modules from private to public crate-level.
rocketmq-client/src/consumer/consumer_impl.rs	Added a new module pull_request_ext.
rocketmq-client/src/consumer/consumer_impl/consume_message_concurrently_service.rs	Enhanced message consumption logic with new fields and methods for handling concurrent message processing.
rocketmq-client/src/consumer/consumer_impl/consume_message_orderly_service.rs	Updated method signatures to improve message handling and memory management.
rocketmq-client/src/consumer/consumer_impl/consume_message_pop_concurrently_service.rs	Similar updates to improve parameter types for message handling.
rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs	Updated method signatures for better message processing.
rocketmq-client/src/consumer/consumer_impl/consume_message_service.rs	Adjusted method signatures to enhance memory management and concurrency.
rocketmq-client/src/consumer/consumer_impl/default_mq_push_consumer_impl.rs	Added new fields and methods to improve message consumption management and error handling.
rocketmq-client/src/consumer/consumer_impl/process_queue.rs	Updated fields and methods to support asynchronous operations and improved message handling.
rocketmq-client/src/consumer/consumer_impl/pull_api_wrapper.rs	Enhanced message processing and broker interaction with new methods and improved error handling.
rocketmq-client/src/consumer/consumer_impl/pull_request.rs	Changed visibility and ownership semantics for PullRequest.
rocketmq-client/src/consumer/consumer_impl/pull_request_ext.rs	Introduced a new struct PullResultExt for extended pull request results.
rocketmq-client/src/consumer/consumer_impl/re_balance.rs	Modified method signatures to allow state changes in rebalance operations.
rocketmq-client/src/consumer/consumer_impl/re_balance/rebalance_impl.rs	Updated fields and methods for improved subscription handling and message queue management.
rocketmq-client/src/consumer/consumer_impl/re_balance/rebalance_push_impl.rs	Enhanced methods for dynamic configuration adjustments based on active message queues.
rocketmq-client/src/consumer/listener/message_listener_concurrently.rs	Updated method signatures to improve message context handling.
rocketmq-client/src/consumer/pull_callback.rs	Introduced new trait and updated callback handling for pull results.
rocketmq-client/src/consumer/pull_result.rs	Changed message list handling to improve message reference management.
rocketmq-client/src/consumer/store/offset_store.rs	Updated reference handling in read_offset method.
rocketmq-client/src/consumer/store/remote_broker_offset_store.rs	Enhanced offset fetching logic with error handling and retry mechanisms.
rocketmq-client/src/factory/mq_client_instance.rs	Adjusted field visibility and method signatures for better access and functionality.
rocketmq-client/src/hook/consume_message_context.rs	Updated message list handling and access channel fields.
rocketmq-client/src/hook/consume_message_hook.rs	Modified hook methods to allow mutable context references.
rocketmq-client/src/hook/filter_message_context.rs	Enhanced flexibility in message context handling with optional fields.
rocketmq-client/src/hook/filter_message_hook.rs	Changed method signature to require a valid context.
rocketmq-client/src/implementation/mq_client_api_impl.rs	Introduced new methods for querying consumer offsets and pulling messages, enhancing client-broker interactions.
rocketmq-client/src/producer/default_mq_producer.rs	Updated method signatures to simplify MessageQueue handling.
rocketmq-common/src/common/base/service_state.rs	Added Display trait implementation for ServiceState enum.
rocketmq-common/src/common/message/message_client_ext.rs	Introduced MessageClientExt struct with various message management methods.
rocketmq-common/src/common/message/message_decoder.rs	Added functions for batch decoding and message property handling.
rocketmq-common/src/common/message/message_queue.rs	Added inline attribute to set_topic method for performance optimization.
rocketmq-common/src/common/message/message_single.rs	Introduced new methods for flexible message creation.
rocketmq-remoting/src/protocol/header.rs	Modified module exports to reorganize public interface.

Assessment against linked issues

Objective	Addressed	Explanation
Support for client clustering consume (#972)	❓	The changes enhance message handling but do not explicitly mention clustering support.

Possibly related PRs

• [ISSUE #658]🚀Support pull message consume #659: Adds support for pull message consumption, enhancing message handling.
• [ISSUE #696]🚀Support pull message consume-5 #697: Focuses on improving pull message consumption, aligning with this PR's changes.
• [ISSUE #699]🚀Support pull message result handle #701: Supports handling pull message results, relevant to the changes in this PR.
• [ISSUE #702]🚀Support pull message result handle-2🚀 #706: Continues to enhance pull message results handling, related to this PR's focus.
• [ISSUE #739]🔥Improve the pull message consumption pattern🚀 #740: Aims to improve pull message consumption patterns, connecting to this PR's modifications.
• [ISSUES #969] Support AdminBrokerProcessor get_all_consumer_offset #970: Introduces functionality to get all consumer offsets, related to overall message management improvements.

Poem

🐰 In the meadow where messages flow,
With each change, our systems grow.
Clusters dance, and offsets align,
A world of messages, all so fine!
Hooray for the code, let it gleam,
In RocketMQ, we build our dream! 🌟

---

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?

Share

• X
• Mastodon
• Reddit
• LinkedIn

Tips

Chat

There are 3 ways to chat with CodeRabbit:

• Review comments: Directly reply to a review comment made by CodeRabbit. Example:
  -- I pushed a fix in commit <commit_id>, please review it.
  -- Generate unit testing code for this file.
  • Open a follow-up GitHub issue for this discussion.
• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
  -- @coderabbitai generate unit testing code for this file.
  -- @coderabbitai modularize this function.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  -- @coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.
  -- @coderabbitai read src/utils.ts and generate unit testing code.
  -- @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.
  -- @coderabbitai help me debug CodeRabbit configuration file.

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)

• @coderabbitai pause to pause the reviews on a PR.
• @coderabbitai resume to resume the paused reviews.
• @coderabbitai review to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
• @coderabbitai full review to do a full review from scratch and review all the files again.
• @coderabbitai summary to regenerate the summary of the PR.
• @coderabbitai resolve resolve all the CodeRabbit review comments.
• @coderabbitai configuration to show the current CodeRabbit configuration for the repository.
• @coderabbitai help to get help.

Other keywords and placeholders

• Add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.
• Add @coderabbitai summary to generate the high-level summary at a specific location in the PR description.
• Add @coderabbitai anywhere in the PR title to generate the title automatically.

Documentation and Community

• Visit our Documentation for detailed information on how to use CodeRabbit.
• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[mxsm]

🚀Thanks for your contribution🎉. CodeRabbit(AI) will review your code first🔥

[mxsm]

🚀Thanks for your contribution🎉. CodeRabbit(AI) will review your code first🔥

[codecov]

Codecov Report

Attention: Patch coverage is 0% with 844 lines in your changes missing coverage. Please review.

Project coverage is 19.84%. Comparing base (887e9f6) to head (ef9380c).
Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
...sumer_impl/consume_message_concurrently_service.rs	0.00%	205 Missing ⚠️
...mer/consumer_impl/default_mq_push_consumer_impl.rs	0.00%	142 Missing ⚠️
...ent/src/consumer/consumer_impl/pull_api_wrapper.rs	0.00%	103 Missing ⚠️
...ketmq-common/src/common/message/message_decoder.rs	0.00%	85 Missing ⚠️
...mq-common/src/common/message/message_client_ext.rs	0.00%	64 Missing ⚠️
...client/src/consumer/consumer_impl/process_queue.rs	0.00%	60 Missing ⚠️
...mq-client/src/implementation/mq_client_api_impl.rs	0.00%	47 Missing ⚠️
...ol/header/consumer_send_msg_back_request_header.rs	0.00%	39 Missing ⚠️
rocketmq-client/src/consumer/pull_callback.rs	0.00%	26 Missing ⚠️
rocketmq-client/src/hook/filter_message_context.rs	0.00%	23 Missing ⚠️
... and 10 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #980      +/-   ##
==========================================
- Coverage   21.02%   19.84%   -1.18%     
==========================================
  Files         408      410       +2     
  Lines       31710    33633    +1923     
==========================================
+ Hits         6667     6676       +9     
- Misses      25043    26957    +1914     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.