[ISSUE #7252]✨enhance(allocate_mapped_file_service, commit_log, local_file_message_store): integrate warm mapped file configuration and improve file allocation logic#7253
Merged
Conversation
…_file_message_store): integrate warm mapped file configuration and improve file allocation logic
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
Collaborator
|
🔊@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
Contributor
WalkthroughIntegrates warm-mapped-file configuration into Changes
Sequence DiagramsequenceDiagram
participant LSM as LocalFileMessageStore
participant AMFS as AllocateMappedFileService
participant CL as CommitLog
participant MFQ as MappedFileQueue
participant MF as MappedFile
LSM->>AMFS: new_with_message_store_config(config)
AMFS->>AMFS: extract WarmMappedFileConfig
LSM->>CL: new(..., allocate_service)
CL->>MFQ: new(..., Some(allocate_service))
MFQ->>MFQ: store allocate_mapped_file_service
rect rgba(100, 150, 200, 0.5)
Note over MFQ: File Allocation Flow
MFQ->>AMFS: is_started()?
alt Service Started
AMFS-->>MFQ: true
MFQ->>AMFS: allocate_mapped_file(...)
AMFS->>MF: create_mapped_file(warm_config)
MF->>MF: warm_mapped_file() if size >= threshold
else Service Not Started
AMFS-->>MFQ: false
MFQ->>MF: DefaultMappedFile::try_new (sync fallback)
end
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning Review ran into problems🔥 ProblemsGit: Failed to clone repository. Please run the 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. Review rate limit: 7/8 reviews remaining, refill in 7 minutes and 30 seconds.Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rocketmq-store/src/base/allocate_mapped_file_service.rs`:
- Around line 404-409: The warm-up call mapped_file.warm_mapped_file is
currently performed before allocation completion is signaled, which blocks the
allocation critical path and can trigger timeouts; change the flow so allocation
completion is signaled before doing expensive warm-up (i.e., move the
mapped_file.warm_mapped_file call to run after the allocation result is
published/notify), or if you must keep warm-up inline, increase the wait budget
for callers when warm_mapped_file_config.should_warm(file_size) is true by
adding a configurable extra timeout; update the logic around
warm_mapped_file_config and the allocation completion/notification code in
allocate_mapped_file_service to reflect this change.
In `@rocketmq-store/src/message_store/local_file_message_store.rs`:
- Around line 457-465: The code freezes transient_store_pool_enable into
allocate_transient_store_pool/allocate_mapped_file_service at construction so a
later sync_broker_role() change won't update allocation behavior; instead remove
the startup-only gate by making AllocateMappedFileService consult the current
role/config at allocation time (e.g., accept a reference/closure to query
broker_role and message_store_config.fast_fail_if_no_buffer_in_store_pool or add
a reconfigure/set_transient_pool_enabled method on AllocateMappedFileService)
and update callers that currently use
AllocateMappedFileService::new_with_message_store_config to either pass a live
check or call the reconfigure setter from sync_broker_role(), ensuring
allocation decisions are made from current broker_role rather than the frozen
transient_store_pool_enable value.
- Around line 4042-4048: Add a focused regression test that asserts
MappedFileQueue creation behavior when the allocate service is started vs not
started: instantiate a test CommitLog/MappedFileQueue from
new_configured_test_store (or directly construct MappedFileQueue) and verify
that when allocate_mapped_file_service().is_started() is true the async
allocation path is used (e.g., file creation is delegated and not performed
synchronously), and when is_started() is false the code falls back to
synchronous creation; use the existing
commit_log.has_allocate_mapped_file_service() check only as setup but add
explicit assertions that exercise the two branches (started and not-started) to
fail if the service-gate behavior in MappedFileQueue or CommitLog (methods on
commit_log / MappedFileQueue and allocate_mapped_file_service) is not honored.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 111ede65-80ad-4fdb-80c4-ce7fec5ac12f
📒 Files selected for processing (4)
rocketmq-store/src/base/allocate_mapped_file_service.rsrocketmq-store/src/consume_queue/mapped_file_queue.rsrocketmq-store/src/log_file/commit_log.rsrocketmq-store/src/message_store/local_file_message_store.rs
Comment on lines
+404
to
+409
| if warm_mapped_file_config.should_warm(file_size) { | ||
| mapped_file.warm_mapped_file( | ||
| warm_mapped_file_config.flush_disk_type, | ||
| warm_mapped_file_config.flush_least_pages_when_warm_mapped_file, | ||
| ); | ||
| } |
Contributor
There was a problem hiding this comment.
Warm-up now sits on the allocation critical path and can cause false timeouts
Line 404 performs warm-up before completion is signaled. With the fixed 5s wait budget, large files can time out callers even though allocation eventually succeeds.
💡 Proposed mitigation (extend wait budget when warm-up is enabled)
const WAIT_TIMEOUT: Duration = Duration::from_secs(5);
+const WAIT_TIMEOUT_WITH_WARM: Duration = Duration::from_secs(30);
impl AllocateMappedFileService {
+ #[inline]
+ fn request_wait_timeout(&self) -> Duration {
+ if self.warm_mapped_file_config.enabled {
+ WAIT_TIMEOUT_WITH_WARM
+ } else {
+ WAIT_TIMEOUT
+ }
+ }
+
pub async fn put_request_and_return_mapped_file(
&self,
next_file_path: String,
next_next_file_path: String,
file_size: i32,
) -> Result<Option<Arc<DefaultMappedFile>>, RocketMQError> {
@@
- let wait_result = tokio::time::timeout(WAIT_TIMEOUT, req.wait()).await;
+ let wait_result = tokio::time::timeout(self.request_wait_timeout(), req.wait()).await;
@@
fn put_request_and_return_mapped_file_blocking(
&self,
next_file_path: String,
next_next_file_path: String,
file_size: i32,
) -> Result<Option<Arc<DefaultMappedFile>>, RocketMQError> {
@@
- if req.wait_blocking(WAIT_TIMEOUT) {
+ if req.wait_blocking(self.request_wait_timeout()) {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-store/src/base/allocate_mapped_file_service.rs` around lines 404 -
409, The warm-up call mapped_file.warm_mapped_file is currently performed before
allocation completion is signaled, which blocks the allocation critical path and
can trigger timeouts; change the flow so allocation completion is signaled
before doing expensive warm-up (i.e., move the mapped_file.warm_mapped_file call
to run after the allocation result is published/notify), or if you must keep
warm-up inline, increase the wait budget for callers when
warm_mapped_file_config.should_warm(file_size) is true by adding a configurable
extra timeout; update the logic around warm_mapped_file_config and the
allocation completion/notification code in allocate_mapped_file_service to
reflect this change.
Comment on lines
+457
to
+465
| let transient_store_pool_enable = message_store_config.transient_store_pool_enable | ||
| && (broker_config.enable_controller_mode || message_store_config.broker_role != BrokerRole::Slave); | ||
| let allocate_transient_store_pool = transient_store_pool_enable.then(|| Arc::new(transient_store_pool.clone())); | ||
| let allocate_mapped_file_service = Arc::new(AllocateMappedFileService::new_with_message_store_config( | ||
| allocate_transient_store_pool, | ||
| transient_store_pool_enable, | ||
| message_store_config.fast_fail_if_no_buffer_in_store_pool, | ||
| message_store_config.as_ref(), | ||
| )); |
Contributor
There was a problem hiding this comment.
Don't freeze allocation-service eligibility at construction time.
transient_store_pool_enable is computed once from the startup role here, but sync_broker_role() can change broker_role later. After a promotion/demotion, the allocate service keeps the old transient-pool/warm-allocation mode, so file allocation can diverge from the store's current role. Recompute/reconfigure this when the role changes, or move the role gate into the allocation path instead of baking it into construction.
Also applies to: 2700-2704
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-store/src/message_store/local_file_message_store.rs` around lines
457 - 465, The code freezes transient_store_pool_enable into
allocate_transient_store_pool/allocate_mapped_file_service at construction so a
later sync_broker_role() change won't update allocation behavior; instead remove
the startup-only gate by making AllocateMappedFileService consult the current
role/config at allocation time (e.g., accept a reference/closure to query
broker_role and message_store_config.fast_fail_if_no_buffer_in_store_pool or add
a reconfigure/set_transient_pool_enabled method on AllocateMappedFileService)
and update callers that currently use
AllocateMappedFileService::new_with_message_store_config to either pass a live
check or call the reconfigure setter from sync_broker_role(), ensuring
allocation decisions are made from current broker_role rather than the frozen
transient_store_pool_enable value.
Comment on lines
+4042
to
+4048
| #[test] | ||
| fn commitlog_uses_allocate_mapped_file_service_for_file_creation() { | ||
| let temp_dir = tempdir().unwrap(); | ||
| let store = new_configured_test_store(&temp_dir, MessageStoreConfig::default()); | ||
|
|
||
| assert!(store.commit_log.has_allocate_mapped_file_service()); | ||
| } |
Contributor
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Cover the new allocation branches, not just service injection.
This test only proves that a service is present. It would still pass if MappedFileQueue always fell back to synchronous creation or never honored the new is_started() gate, which is the behavior this PR actually changes. Please add a focused regression test for the started vs. not-started paths.
As per coding guidelines "For behavior changes, add or update focused tests that would fail without the change."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-store/src/message_store/local_file_message_store.rs` around lines
4042 - 4048, Add a focused regression test that asserts MappedFileQueue creation
behavior when the allocate service is started vs not started: instantiate a test
CommitLog/MappedFileQueue from new_configured_test_store (or directly construct
MappedFileQueue) and verify that when
allocate_mapped_file_service().is_started() is true the async allocation path is
used (e.g., file creation is delegated and not performed synchronously), and
when is_started() is false the code falls back to synchronous creation; use the
existing commit_log.has_allocate_mapped_file_service() check only as setup but
add explicit assertions that exercise the two branches (started and not-started)
to fail if the service-gate behavior in MappedFileQueue or CommitLog (methods on
commit_log / MappedFileQueue and allocate_mapped_file_service) is not honored.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7253 +/- ##
==========================================
+ Coverage 60.18% 60.21% +0.03%
==========================================
Files 1093 1093
Lines 203701 203794 +93
==========================================
+ Hits 122588 122714 +126
+ Misses 81113 81080 -33 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
approved these changes
May 1, 2026
rocketmq-rust-bot
left a comment
rocketmq-rust-bot
left a comment
Collaborator
There was a problem hiding this comment.
LGTM - All CI checks passed ✅
rocketmq-rust-bot
added
approved
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which Issue(s) This PR Fixes(Closes)
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
New Features
Bug Fixes