[ISSUE #9343]♻️Inject remoting command factories into Broker request processors

[mxsm]

Which Issue(s) This PR Fixes(Closes)

• Fixes [Refactor♻️] Inject remoting command factories into Broker request processors #9343

Brief Description

Inject the Broker-owned RemotingCommandFactory through runtime composition, request processor contexts, Broker2Client, and fast-failure paths so each Broker instance controls command version and serialization defaults. Preserve compatibility constructors for external callers, route transport error and RPC response helpers through explicit factory-aware APIs, and add a production-source audit plus JSON/ROCKETMQ owner-default regression coverage.

How Did You Test This Change?

• cargo fmt -p <workspace-package> -- --check for all 28 root workspace packages (Windows-safe equivalent after cargo fmt --all -- --check hit OS error 206)
• cargo clippy --workspace --no-deps --all-targets --all-features -- -D warnings
• cargo test -p rocketmq-broker --test remoting_command_factory_owner_audit
• cargo test -p rocketmq-broker consume_message_directly_returns_offline_error_for_known_message --lib
• cargo test -p rocketmq-broker invoke_broker_to_get_consumer_status_returns_offline_group_error --lib
• cargo test -p rocketmq-broker invoke_broker_to_reset_offset_preserves_factory --lib
• cargo test -p rocketmq-broker --lib -- --skip processor::send_message_processor::capability::tests::extended_timer_capability_is_published_from_one_policy_generation
• cargo test -p rocketmq-transport --lib
• cargo test -p rocketmq-protocol --test remoting_wire_golden
• cargo doc -p rocketmq-broker -p rocketmq-transport --no-deps
• cargo +nightly-2026-07-05 check --locked --all-targets --all-features from fuzz/
• Standalone rocketmq-example: per-package format check and cargo clippy --all-targets -- -D warnings
• Standalone rocketmq-mcp: per-package format check, read-only boundary check, locked check/tests/all-feature tests, Streamable HTTP Clippy, and docs
• .\scripts\runtime-audit.ps1 -SkipBaseline -EnforceBoundaryBaseline
• .\scripts\check-error-hygiene.ps1
• git diff --check

The unmodified Broker baseline test processor::send_message_processor::capability::tests::extended_timer_capability_is_published_from_one_policy_generation fails identically on main; the remaining 803 Broker library tests pass when that known baseline failure is skipped.

Summary by CodeRabbit

• New Features
  • Added configurable remoting command creation across broker startup, runtime processing, client communication, and RPC responses.
  • Added factory-aware APIs for customizing response formats and protocol defaults.
  • Existing constructors continue using application defaults for compatibility.
• Tests
  • Added coverage verifying independent configurations, response behavior, and prevention of bypassing configured command creation.

[rocketmq-rust-bot]

🔊@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💥.

[coderabbitai]

Caution

Review failed

The pull request is closed.

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro Plus

Run ID: 92056658-06e7-41e0-9377-ea2e83b61daf

📥 Commits

Reviewing files that changed from the base of the PR and between a69d914 and b854fee.

📒 Files selected for processing (39)

• rocketmq-broker/src/broker_bootstrap.rs
• rocketmq-broker/src/broker_runtime.rs
• rocketmq-broker/src/broker_runtime/composition.rs
• rocketmq-broker/src/broker_runtime/data_plane.rs
• rocketmq-broker/src/broker_runtime/request_pipeline.rs
• rocketmq-broker/src/client/net/broker_to_client.rs
• rocketmq-broker/src/latency/broker_fast_failure.rs
• rocketmq-broker/src/processor.rs
• rocketmq-broker/src/processor/ack_message_processor.rs
• rocketmq-broker/src/processor/admin_broker_processor.rs
• rocketmq-broker/src/processor/admin_broker_processor/consumer_request_handler.rs
• rocketmq-broker/src/processor/change_invisible_time_processor.rs
• rocketmq-broker/src/processor/client_manage_processor.rs
• rocketmq-broker/src/processor/consumer_manage_processor.rs
• rocketmq-broker/src/processor/default_pull_message_result_handler.rs
• rocketmq-broker/src/processor/end_transaction_processor.rs
• rocketmq-broker/src/processor/lite_manager_processor.rs
• rocketmq-broker/src/processor/lite_subscription_ctl_processor.rs
• rocketmq-broker/src/processor/maintenance_request_processor.rs
• rocketmq-broker/src/processor/notification_processor.rs
• rocketmq-broker/src/processor/peek_message_processor.rs
• rocketmq-broker/src/processor/polling_info_processor.rs
• rocketmq-broker/src/processor/pop_lite_message_processor.rs
• rocketmq-broker/src/processor/pop_message_processor.rs
• rocketmq-broker/src/processor/pop_message_processor/capability.rs
• rocketmq-broker/src/processor/pull_message_processor.rs
• rocketmq-broker/src/processor/pull_message_processor/capability.rs
• rocketmq-broker/src/processor/query_assignment_processor.rs
• rocketmq-broker/src/processor/query_message_processor.rs
• rocketmq-broker/src/processor/recall_message_processor.rs
• rocketmq-broker/src/processor/reply_message_processor.rs
• rocketmq-broker/src/processor/send_message_processor.rs
• rocketmq-broker/src/processor/send_message_processor/capability.rs
• rocketmq-broker/src/proxy_facade.rs
• rocketmq-broker/src/topic/manager/topic_queue_mapping_manager.rs
• rocketmq-broker/tests/remoting_command_factory_owner_audit.rs
• rocketmq-transport/src/error_response.rs
• rocketmq-transport/src/public_api.rs
• rocketmq-transport/src/rpc/rpc_client_utils.rs

---

Walkthrough

The pull request injects RemotingCommandFactory through broker construction, runtime state, request processors, broker-to-client operations, fast-failure handling, static-topic rewriting, and transport response helpers. Existing constructors keep application-default behavior.

Changes

Broker factory injection

Layer / File(s)	Summary
Runtime factory ownership and wiring rocketmq-broker/src/broker_bootstrap.rs, rocketmq-broker/src/broker_runtime/..., rocketmq-broker/src/latency/broker_fast_failure.rs	Builders accept custom factories. Runtime state stores the selected factory. Request pipelines and fast-failure queues receive it.
Broker clients and dispatch responses rocketmq-broker/src/client/net/broker_to_client.rs, rocketmq-broker/src/processor.rs	Broker-to-client requests and broker dispatch responses use stored factories. Default constructors preserve application defaults.
Core message and transaction processors rocketmq-broker/src/processor/{ack_message_processor,admin_broker_processor,change_invisible_time_processor,client_manage_processor,consumer_manage_processor,end_transaction_processor}.rs, rocketmq-broker/src/processor/admin_broker_processor/consumer_request_handler.rs	Processor contexts and handlers propagate factories through success, error, forwarding, and validation responses.
Lite, polling, POP, pull, and query processors rocketmq-broker/src/processor/*	Factory-aware response creation covers lite, polling, POP, pull, query-assignment, query-message, recall, and reply processing, including static-topic rewrites.
Send flow and transport helpers rocketmq-broker/src/processor/send_message_processor*, rocketmq-broker/src/proxy_facade.rs, rocketmq-broker/src/topic/manager/topic_queue_mapping_manager.rs, rocketmq-transport/src/{error_response.rs,public_api.rs,rpc/rpc_client_utils.rs}, rocketmq-broker/tests/remoting_command_factory_owner_audit.rs	Send-message and proxy construction pass explicit factories. Transport helpers expose factory-aware error and RPC response APIs. Tests verify wire defaults and reject bypasses.

Estimated code review effort: 4 (Complex) | ~60 minutes

Possibly related PRs

• mxsm/rocketmq-rust#9298: Introduces application-wide remoting-command defaults used by this propagation work.
• mxsm/rocketmq-rust#9333: Propagates RemotingCommandFactory through broker runtime composition.
• mxsm/rocketmq-rust#9340: Injects owner-scoped factories through broker and NameServer construction paths.

Poem

A rabbit carried commands through the broker bright,
With JSON and RocketMQ defaults set right.
Each processor kept its chosen tone,
No static constructor stood alone.
“Hop,” said the rabbit, “the factory now flows!”

✨ Finishing Touches 💡 1

🛠️ Fix failing CI checks 💡

• Create stacked PR
• Commit on current branch

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch mxsm/issue-9343-broker-processor-command-factory

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}