Repository navigation
[ISSUE #7069]🚀feat(commit-log): implement file deletion services for expired commit logs and consume queues, enhance offset correction logic, and add tests for cleanup functionality - #7070
Merged
Conversation
…expired commit logs and consume queues, enhance offset correction logic, and add tests for cleanup functionality
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
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
April 5, 2026 14:37
Contributor
WalkthroughThis pull request implements file deletion and cleanup services for expired commit logs and consume queues, refactors service initialization with proper dependency injection, adds offset correction logic, and introduces comprehensive test coverage for cleanup operations. Changes
Sequence Diagram(s)sequenceDiagram
participant MessageStore
participant CleanCommitLogService
participant CommitLog
participant MappedFileQueue
participant IndexService
MessageStore->>CleanCommitLogService: run()
CleanCommitLogService->>CleanCommitLogService: check atomic manual_clean flag
CleanCommitLogService->>CleanCommitLogService: compute expiration_time from file_reserved_time
CleanCommitLogService->>CommitLog: delete_expired_files_by_time(expired_time, ...)
CommitLog->>MappedFileQueue: delete_expired_files_by_time(...)
MappedFileQueue-->>CommitLog: return deleted_count
CommitLog-->>CleanCommitLogService: return deleted_count
alt deleted_count > 0
CleanCommitLogService->>CleanCommitLogService: log deletion event
end
CleanCommitLogService->>CommitLog: retry_delete_first_file(interval_forcibly)
CommitLog->>MappedFileQueue: retry_delete_first_file(...)
MappedFileQueue-->>CommitLog: return retry_result
CommitLog-->>CleanCommitLogService: return retry_result
sequenceDiagram
participant MessageStore
participant CorrectLogicOffsetService
participant CommitLog
participant ConsumeQueueStore
MessageStore->>CorrectLogicOffsetService: run()
CorrectLogicOffsetService->>CommitLog: get_min_offset()
CommitLog-->>CorrectLogicOffsetService: min_commit_log_offset
alt min_commit_log_offset >= 0
CorrectLogicOffsetService->>ConsumeQueueStore: iterate consume_queue_tables
loop for each consume_queue
CorrectLogicOffsetService->>ConsumeQueueStore: correct_min_offset(min_commit_log_offset)
end
else
CorrectLogicOffsetService->>CorrectLogicOffsetService: exit early
end
sequenceDiagram
participant MessageStore
participant CleanConsumeQueueService
participant ConsumeQueueStore
participant CommitLog
participant IndexService
MessageStore->>CleanConsumeQueueService: run()
CleanConsumeQueueService->>CommitLog: get_min_offset()
CommitLog-->>CleanConsumeQueueService: min_commit_log_offset
alt min_commit_log_offset < 0
CleanConsumeQueueService->>CleanConsumeQueueService: exit early
else
loop for each consume_queue
CleanConsumeQueueService->>ConsumeQueueStore: delete_expired_files(min_commit_log_offset)
CleanConsumeQueueService->>ConsumeQueueStore: correct_min_offset(min_commit_log_offset)
end
CleanConsumeQueueService->>IndexService: delete_expired_file()
CleanConsumeQueueService->>ConsumeQueueStore: clean_expired(min_commit_log_offset)
alt Tokio runtime detected
ConsumeQueueStore-->>CleanConsumeQueueService: run synchronously
else
ConsumeQueueStore->>ConsumeQueueStore: spawn thread
ConsumeQueueStore-->>CleanConsumeQueueService: join thread
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
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. 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: 5
🤖 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/message_store/local_file_message_store.rs`:
- Around line 2853-2891: The public trait method
MessageStore::clean_expired_consumer_queue currently calls only
consume_queue_store.clean_expired(...) and therefore skips the new steps added
in LocalFileMessageStore::run() (delete_expired_file, correct_min_offset
adjustments and index_service pruning); update the trait implementation for
clean_expired_consumer_queue (or the implementation on LocalFileMessageStore) to
route through the same logic as run(): either call a new shared helper that
performs the sequence (iterate consume queues -> delete_expired_file ->
correct_min_offset -> index_service.delete_expired_file -> await
consume_queue_store.clean_expired) or invoke the existing run() cleanup path so
callers using the trait get the same behavior; reference symbols:
LocalFileMessageStore::run, MessageStore::clean_expired_consumer_queue,
consume_queue_store.delete_expired_file, consume_queue_store.correct_min_offset,
index_service.delete_expired_file, and consume_queue_store.clean_expired.
- Around line 2812-2819: The code currently collapses all queued manual deletes
into a single boolean via self.manual_delete_requests.swap(0) > 0; instead
capture the actual count (let mut manual_count =
self.manual_delete_requests.swap(0, Ordering::SeqCst)) and use that count to
drive deletion until the backlog is drained: call
commit_log.mut_from_ref().delete_expired_files_by_time(...) in a loop while
manual_count > 0, passing clean_immediately = true for those manual-driven
iterations, subtract the number of files actually deleted (the function's
return) from manual_count each iteration, and break if the delete call returns
0; this preserves execute_delete_files_manually() increments and ensures
delete_file_batch_max doesn't silently discard extra manual requests in run().
- Around line 248-260: Schedule commit-log deletion and CQ/index correction to
run in the same scheduled pass: modify add_schedule_task so it does not spawn
clean_commit_log_service in one Tokio task and
correct_logic_offset_service/clean_consume_queue_service in another, but instead
executes CleanCommitLogService::clean (or the existing clean_commit_log_service
task) first and then immediately runs CorrectLogicOffsetService::correct and
CleanConsumeQueueService::clean within the same scheduled closure/future; this
ensures the consume-queue/index corrections observe the final commit-log min
offset and prevents them from seeing and returning on stale offsets between
separate Tokio tasks.
In `@rocketmq-store/src/queue/local_file_consume_queue_store.rs`:
- Around line 239-241: The queue_offset_operator.remove(...) call is unsafe
because when a consume queue is trimmed-to-empty and later recreated via
find_or_create_consume_queue(), its offset entry is lost and get_queue_offset()
will default to 0, causing offset reuse; change the logic around the remove to
either skip deleting the offset operator for trimmed-to-empty queues or ensure
the offset operator is restored on recreation by initializing
queue_offset_operator with the consume_queue's persisted max offset (use
consume_queue.get_max_offset() or equivalent) inside
find_or_create_consume_queue(); specifically, modify the code path that
currently calls queue_offset_operator.remove() (the block using
consume_queue.get_message_total_in_queue() and get_min_offset_in_queue()) to
preserve the entry, or add code in find_or_create_consume_queue() to repopulate
queue_offset_operator from the recreated consume_queue's persisted state so
offsets continue monotonically.
In `@rocketmq-store/src/queue/single_consume_queue.rs`:
- Around line 930-955: The loop that recomputes min_logic_offset must ignore
pre-blank filler records (written by fill_pre_blank as (0, i32::MAX, 0)); update
the conditional that currently checks `if offset_py >= min_commit_log_offset &&
size > 0` in the loop inside the mapped file iteration to also exclude filler
entries (e.g. require `size != i32::MAX` and/or `offset_py != 0`), so filler
slots are skipped (continue) and never used to store `min_logic_offset`,
`min_ext_addr`, or trigger `consume_queue_ext.truncate_by_min_address`; locate
the check around `offset_py`, `size`, `tags_code`, `min_logic_offset`, and
`consume_queue_ext` and add the extra guard there.
🪄 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: a425cdb6-df2c-499b-b14d-bc3092363879
📒 Files selected for processing (4)
rocketmq-store/src/log_file/commit_log.rsrocketmq-store/src/message_store/local_file_message_store.rsrocketmq-store/src/queue/local_file_consume_queue_store.rsrocketmq-store/src/queue/single_consume_queue.rs
Comment on lines
+248
to
+260
| clean_commit_log_service: Arc::new(CleanCommitLogService::new( | ||
| message_store_config.clone(), | ||
| commit_log.clone(), | ||
| )), | ||
| correct_logic_offset_service: Arc::new(CorrectLogicOffsetService::new( | ||
| commit_log.clone(), | ||
| consume_queue_store.clone(), | ||
| )), | ||
| clean_consume_queue_service: Arc::new(CleanConsumeQueueService::new( | ||
| commit_log.clone(), | ||
| consume_queue_store.clone(), | ||
| index_service.clone(), | ||
| )), |
Contributor
There was a problem hiding this comment.
Keep commit-log deletion and CQ/index correction in the same scheduled pass.
add_schedule_task() runs clean_commit_log_service in one Tokio loop and correct_logic_offset_service/clean_consume_queue_service in another. After this split, the CQ task can observe the old commit-log min offset, return, and then the commit-log task advances it, so consume queues and indexes can still point at deleted commit-log files until the next cleanup interval.
🤖 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
248 - 260, Schedule commit-log deletion and CQ/index correction to run in the
same scheduled pass: modify add_schedule_task so it does not spawn
clean_commit_log_service in one Tokio task and
correct_logic_offset_service/clean_consume_queue_service in another, but instead
executes CleanCommitLogService::clean (or the existing clean_commit_log_service
task) first and then immediately runs CorrectLogicOffsetService::correct and
CleanConsumeQueueService::clean within the same scheduled closure/future; this
ensures the consume-queue/index corrections observe the final commit-log min
offset and prevents them from seeing and returning on stale offsets between
separate Tokio tasks.
Comment on lines
+2812
to
+2819
| let clean_immediately = self.manual_delete_requests.swap(0, Ordering::SeqCst) > 0; | ||
| let delete_count = self.commit_log.mut_from_ref().delete_expired_files_by_time( | ||
| expired_time, | ||
| self.message_store_config.delete_commit_log_files_interval as i32, | ||
| self.message_store_config.destroy_mapped_file_interval_forcibly as i64, | ||
| clean_immediately, | ||
| self.message_store_config.delete_file_batch_max as i32, | ||
| ); |
Contributor
There was a problem hiding this comment.
Don't collapse queued manual deletions into a single boolean.
execute_delete_files_manually() increments a counter, but run() reduces it to swap(0) > 0. If delete_file_batch_max only clears part of the backlog in one pass, every extra manual request is discarded and later runs stop forcing deletion even though files are still pending.
Suggested fix
- let clean_immediately = self.manual_delete_requests.swap(0, Ordering::SeqCst) > 0;
+ let clean_immediately = self
+ .manual_delete_requests
+ .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |count| {
+ (count > 0).then_some(count - 1)
+ })
+ .is_ok();Also applies to: 2833-2835
🤖 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
2812 - 2819, The code currently collapses all queued manual deletes into a
single boolean via self.manual_delete_requests.swap(0) > 0; instead capture the
actual count (let mut manual_count = self.manual_delete_requests.swap(0,
Ordering::SeqCst)) and use that count to drive deletion until the backlog is
drained: call commit_log.mut_from_ref().delete_expired_files_by_time(...) in a
loop while manual_count > 0, passing clean_immediately = true for those
manual-driven iterations, subtract the number of files actually deleted (the
function's return) from manual_count each iteration, and break if the delete
call returns 0; this preserves execute_delete_files_manually() increments and
ensures delete_file_batch_max doesn't silently discard extra manual requests in
run().
Comment on lines
2853
to
+2891
| fn run(&self) { | ||
| error!("clean consume queue service run unimplemented!") | ||
| let min_commit_log_offset = self.commit_log.get_min_offset(); | ||
| if min_commit_log_offset < 0 { | ||
| return; | ||
| } | ||
|
|
||
| let consume_queue_table = self.consume_queue_store.get_consume_queue_table().lock().clone(); | ||
| for queue_table in consume_queue_table.values() { | ||
| for consume_queue in queue_table.values() { | ||
| let consume_queue = &***consume_queue; | ||
| let _ = self | ||
| .consume_queue_store | ||
| .delete_expired_file(consume_queue, min_commit_log_offset); | ||
| self.consume_queue_store | ||
| .correct_min_offset(consume_queue, min_commit_log_offset); | ||
| } | ||
| } | ||
|
|
||
| self.index_service | ||
| .delete_expired_file(min_commit_log_offset.max(0) as u64); | ||
|
|
||
| let consume_queue_store = self.consume_queue_store.clone(); | ||
| let clean_expired = move || { | ||
| let runtime = tokio::runtime::Builder::new_current_thread() | ||
| .enable_all() | ||
| .build() | ||
| .expect("clean consume queue runtime should build"); | ||
| runtime.block_on(async move { | ||
| consume_queue_store.clean_expired(min_commit_log_offset).await; | ||
| }); | ||
| }; | ||
|
|
||
| if Handle::try_current().is_ok() { | ||
| thread::spawn(clean_expired) | ||
| .join() | ||
| .expect("clean consume queue thread should complete"); | ||
| } else { | ||
| clean_expired(); | ||
| } |
Contributor
There was a problem hiding this comment.
Route the public CQ cleanup path through this implementation.
This run() path now deletes expired CQ files, corrects min offsets, and prunes index entries, but MessageStore::clean_expired_consumer_queue() still only calls consume_queue_store.clean_expired(...). Any caller that goes through the trait method will miss the new correction/index cleanup work and keep the pre-PR behavior.
🤖 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
2853 - 2891, The public trait method MessageStore::clean_expired_consumer_queue
currently calls only consume_queue_store.clean_expired(...) and therefore skips
the new steps added in LocalFileMessageStore::run() (delete_expired_file,
correct_min_offset adjustments and index_service pruning); update the trait
implementation for clean_expired_consumer_queue (or the implementation on
LocalFileMessageStore) to route through the same logic as run(): either call a
new shared helper that performs the sequence (iterate consume queues ->
delete_expired_file -> correct_min_offset -> index_service.delete_expired_file
-> await consume_queue_store.clean_expired) or invoke the existing run() cleanup
path so callers using the trait get the same behavior; reference symbols:
LocalFileMessageStore::run, MessageStore::clean_expired_consumer_queue,
consume_queue_store.delete_expired_file, consume_queue_store.correct_min_offset,
index_service.delete_expired_file, and consume_queue_store.clean_expired.
Comment on lines
+239
to
+241
| let message_total_in_queue = consume_queue.get_message_total_in_queue(); | ||
| let queue_trimmed_to_empty = | ||
| message_total_in_queue <= 0 && consume_queue.get_min_offset_in_queue() > 0; |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
QO_FILE="$(fd 'queue_offset_operator\.rs$' rocketmq-store/src | head -n1)"
test -n "$QO_FILE"
printf '%s\n' "--- QueueOffsetOperator methods ---"
rg -n -C4 '\bfn\s+(get_queue_offset|remove|current_queue_offset|set_topic_queue_table|set_lmq_topic_queue_table|increase_queue_offset)\b' "$QO_FILE"
printf '\n%s\n' "--- Cleanup / recreate call sites ---"
rg -n -C3 'queue_trimmed_to_empty|queue_offset_operator\.remove|get_queue_offset|assign_queue_offset|increase_queue_offset|find_or_create_consume_queue' \
rocketmq-store/src/queue/local_file_consume_queue_store.rs \
rocketmq-store/src/queue/single_consume_queue.rsRepository: mxsm/rocketmq-rust
Length of output: 13681
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Get the full find_or_create_consume_queue implementation
rg -n -A 30 'fn find_or_create_consume_queue' rocketmq-store/src/queue/local_file_consume_queue_store.rs | head -50
# Check if there's any code that sets offset_operator when loading/creating queues
rg -n -B2 -A2 'set_topic_queue_table|set_queue_offset|restore.*offset|load.*queue' rocketmq-store/src/queue/local_file_consume_queue_store.rs
# Check the full clean_expired method to see the full context
rg -n -A 60 'fn clean_expired' rocketmq-store/src/queue/local_file_consume_queue_store.rs | head -100Repository: mxsm/rocketmq-rust
Length of output: 6856
Remove the offset operator entry only after confirming the queue will not be recreated, or preserve it across destroy-recreate cycles.
The queue_offset_operator.remove() call at line 293 is unsafe. When a trimmed-to-empty queue is destroyed and later recreated via find_or_create_consume_queue(), the offset operator entry is lost. Since get_queue_offset() defaults missing keys to 0 (via or_insert(0) at queue_offset_operator.rs:51), the next message appended to the recreated queue will be assigned logical offset 0 instead of continuing from the previous maximum—causing logical offset reuse and corrupting the queue's offset sequence.
find_or_create_consume_queue() has no mechanism to restore the offset operator entry on recreation. Unlike the startup path (load() → set_topic_queue_table()), runtime queue recreation does not repopulate the offset operator.
Either preserve the offset operator entry when destroying trimmed-to-empty queues, or restore it during queue recreation using the queue's persisted max offset state.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-store/src/queue/local_file_consume_queue_store.rs` around lines 239
- 241, The queue_offset_operator.remove(...) call is unsafe because when a
consume queue is trimmed-to-empty and later recreated via
find_or_create_consume_queue(), its offset entry is lost and get_queue_offset()
will default to 0, causing offset reuse; change the logic around the remove to
either skip deleting the offset operator for trimmed-to-empty queues or ensure
the offset operator is restored on recreation by initializing
queue_offset_operator with the consume_queue's persisted max offset (use
consume_queue.get_max_offset() or equivalent) inside
find_or_create_consume_queue(); specifically, modify the code path that
currently calls queue_offset_operator.remove() (the block using
consume_queue.get_message_total_in_queue() and get_min_offset_in_queue()) to
preserve the entry, or add code in find_or_create_consume_queue() to repopulate
queue_offset_operator from the recreated consume_queue's persisted state so
offsets continue monotonically.
Comment on lines
+930
to
+955
| let mapped_files = self.mapped_file_queue.get_mapped_files().load().clone(); | ||
| for mapped_file in mapped_files.iter() { | ||
| let read_position = mapped_file.get_read_position(); | ||
| let mut pos = 0; | ||
| while pos + CQ_STORE_UNIT_SIZE <= read_position { | ||
| let Some(mut bytes) = mapped_file.get_bytes(pos as usize, CQ_STORE_UNIT_SIZE as usize) else { | ||
| break; | ||
| } | ||
| let mid = (low + high) / 2 / CQ_STORE_UNIT_SIZE * CQ_STORE_UNIT_SIZE; | ||
| let commit_log_offset = mapped.get_bytes(mid as usize, 8).unwrap().get_i64(); | ||
|
|
||
| match commit_log_offset.cmp(&min_commit_log_offset) { | ||
| std::cmp::Ordering::Greater => high = mid, | ||
| std::cmp::Ordering::Equal => { | ||
| low = mid; | ||
| high = mid; | ||
| break; | ||
| } | ||
| std::cmp::Ordering::Less => low = mid, | ||
| } | ||
| } | ||
| let mut i = low; | ||
| while i <= high { | ||
| let offset_py = mapped.get_bytes(i as usize, 8).unwrap().get_i64(); | ||
| let tags_code = mapped.get_bytes((i + 12) as usize, 8).unwrap().get_i64(); | ||
| if offset_py >= min_commit_log_offset { | ||
| self.min_logic_offset.store( | ||
| mapped.get_file_from_offset() as i64 + i as i64 + start, | ||
| Ordering::SeqCst, | ||
| ); | ||
| }; | ||
| let offset_py = bytes.get_i64(); | ||
| let size = bytes.get_i32(); | ||
| let tags_code = bytes.get_i64(); | ||
| if offset_py >= min_commit_log_offset && size > 0 { | ||
| self.min_logic_offset | ||
| .store(mapped_file.get_file_from_offset() as i64 + pos as i64, Ordering::SeqCst); | ||
| if ConsumeQueue::<MS>::is_ext_addr(tags_code) { | ||
| min_ext_addr = tags_code; | ||
| } | ||
| break; | ||
| if self.is_ext_read_enable() { | ||
| self.consume_queue_ext | ||
| .as_ref() | ||
| .unwrap() | ||
| .truncate_by_min_address(min_ext_addr); | ||
| } | ||
| return; | ||
| } | ||
| i += CQ_STORE_UNIT_SIZE; | ||
| pos += CQ_STORE_UNIT_SIZE; |
Contributor
There was a problem hiding this comment.
Skip pre-blank filler when recomputing min_logic_offset.
fill_pre_blank() writes (0, i32::MAX, 0) padding before the first real slot. Because this loop restarts from pos = 0 and only checks size > 0, a queue whose first real logical offset is non-zero gets clamped back to 0 whenever min_commit_log_offset is 0, exposing filler records as valid entries.
Proposed fix
let mut min_ext_addr = 1i64;
+ let current_min_logic_offset = self.get_min_logic_offset();
let mapped_files = self.mapped_file_queue.get_mapped_files().load().clone();
for mapped_file in mapped_files.iter() {
let read_position = mapped_file.get_read_position();
- let mut pos = 0;
+ let mut pos = if current_min_logic_offset > mapped_file.get_file_from_offset() as i64 {
+ (current_min_logic_offset - mapped_file.get_file_from_offset() as i64) as i32
+ } else {
+ 0
+ };
while pos + CQ_STORE_UNIT_SIZE <= read_position {
let Some(mut bytes) = mapped_file.get_bytes(pos as usize, CQ_STORE_UNIT_SIZE as usize) else {
break;
};
let offset_py = bytes.get_i64();
let size = bytes.get_i32();
let tags_code = bytes.get_i64();
+ if offset_py == 0 && size == i32::MAX {
+ pos += CQ_STORE_UNIT_SIZE;
+ continue;
+ }
if offset_py >= min_commit_log_offset && size > 0 {
self.min_logic_offset
.store(mapped_file.get_file_from_offset() as i64 + pos as i64, Ordering::SeqCst);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-store/src/queue/single_consume_queue.rs` around lines 930 - 955, The
loop that recomputes min_logic_offset must ignore pre-blank filler records
(written by fill_pre_blank as (0, i32::MAX, 0)); update the conditional that
currently checks `if offset_py >= min_commit_log_offset && size > 0` in the loop
inside the mapped file iteration to also exclude filler entries (e.g. require
`size != i32::MAX` and/or `offset_py != 0`), so filler slots are skipped
(continue) and never used to store `min_logic_offset`, `min_ext_addr`, or
trigger `consume_queue_ext.truncate_by_min_address`; locate the check around
`offset_py`, `size`, `tags_code`, `min_logic_offset`, and `consume_queue_ext`
and add the extra guard there.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7070 +/- ##
==========================================
+ Coverage 57.94% 58.07% +0.12%
==========================================
Files 1080 1080
Lines 182101 182350 +249
==========================================
+ Hits 105523 105891 +368
+ Misses 76578 76459 -119 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
approved these changes
Apr 5, 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
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
Release Notes
New Features
Bug Fixes & Improvements