This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 88846a0a5b [ISSUE #10288] Hold MappedFile reference in ConsumeQueue
write path to prevent SIGSEGV (#10566)
88846a0a5b is described below
commit 88846a0a5b550116d183f62c1a468e75ca3f61d4
Author: Jiahua Wang <[email protected]>
AuthorDate: Thu Sep 3 15:55:22 2026 +0800
[ISSUE #10288] Hold MappedFile reference in ConsumeQueue write path to
prevent SIGSEGV (#10566)
ConsumeQueue.putMessagePositionInfo() accesses MappedFile without
holding a reference count. A concurrent cleanExpiredConsumeQueue
can destroy() the MappedFile (unmapping its buffer) while the
writer is still using it, causing SIGSEGV in JIT-compiled
Unsafe.copyMemory.
Fix: Call mappedFile.hold() before accessing the buffer and
release() in a finally block, following the same pattern used
in DefaultMappedFile.getData(). Also protect the fillPreBlank
call in initializeWithOffset.
Reported crash signatures:
- Case 1 (v4.9.4): SEGV_MAPERR in ReputMessageService thread
at ConsumeQueue.putMessagePositionInfo
- Case 2 (v5.3.3): SEGV in AdminBrokerThread at
ConsumeQueueStore.cleanExpired
Co-authored-by: wangjiahua.wjh <[email protected]>
---
.../org/apache/rocketmq/store/ConsumeQueue.java | 86 +++++++++++++---------
1 file changed, 53 insertions(+), 33 deletions(-)
diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
index 0d698dacfe..3503d8fcfc 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
@@ -856,44 +856,52 @@ public class ConsumeQueue implements
ConsumeQueueInterface {
MappedFile mappedFile =
this.mappedFileQueue.getLastMappedFile(expectLogicOffset);
if (mappedFile != null) {
-
- if (mappedFile.isFirstCreateInQueue() && cqOffset != 0 &&
mappedFile.getWrotePosition() == 0) {
- this.minLogicOffset = expectLogicOffset;
- this.mappedFileQueue.setFlushedWhere(expectLogicOffset);
- this.mappedFileQueue.setCommittedWhere(expectLogicOffset);
- this.fillPreBlank(mappedFile, expectLogicOffset);
- log.info("fill pre blank space " + mappedFile.getFileName() +
" " + expectLogicOffset + " "
- + mappedFile.getWrotePosition());
+ if (!mappedFile.hold()) {
+ log.warn("Failed to hold mapped file for ConsumeQueue write,
topic={} queueId={}",
+ this.topic, this.queueId);
+ return false;
}
+ try {
+ if (mappedFile.isFirstCreateInQueue() && cqOffset != 0 &&
mappedFile.getWrotePosition() == 0) {
+ this.minLogicOffset = expectLogicOffset;
+ this.mappedFileQueue.setFlushedWhere(expectLogicOffset);
+ this.mappedFileQueue.setCommittedWhere(expectLogicOffset);
+ this.fillPreBlank(mappedFile, expectLogicOffset);
+ log.info("fill pre blank space " +
mappedFile.getFileName() + " " + expectLogicOffset + " "
+ + mappedFile.getWrotePosition());
+ }
- if (cqOffset != 0) {
- long currentLogicOffset = mappedFile.getWrotePosition() +
mappedFile.getFileFromOffset();
+ if (cqOffset != 0) {
+ long currentLogicOffset = mappedFile.getWrotePosition() +
mappedFile.getFileFromOffset();
- if (expectLogicOffset < currentLogicOffset) {
- log.warn("Build consume queue repeatedly,
expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
- expectLogicOffset, currentLogicOffset, this.topic,
this.queueId, expectLogicOffset - currentLogicOffset);
- return true;
- }
+ if (expectLogicOffset < currentLogicOffset) {
+ log.warn("Build consume queue repeatedly,
expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
+ expectLogicOffset, currentLogicOffset, this.topic,
this.queueId, expectLogicOffset - currentLogicOffset);
+ return true;
+ }
- if (expectLogicOffset != currentLogicOffset) {
- LOG_ERROR.warn(
- "[BUG]logic queue order maybe wrong,
expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
- expectLogicOffset,
- currentLogicOffset,
- this.topic,
- this.queueId,
- expectLogicOffset - currentLogicOffset
- );
+ if (expectLogicOffset != currentLogicOffset) {
+ LOG_ERROR.warn(
+ "[BUG]logic queue order maybe wrong,
expectLogicOffset: {} currentLogicOffset: {} Topic: {} QID: {} Diff: {}",
+ expectLogicOffset,
+ currentLogicOffset,
+ this.topic,
+ this.queueId,
+ expectLogicOffset - currentLogicOffset
+ );
+ }
}
+ this.setMaxPhysicOffset(offset + size);
+ boolean appendResult;
+ if
(messageStore.getMessageStoreConfig().isPutConsumeQueueDataByFileChannel()) {
+ appendResult =
mappedFile.appendMessageUsingFileChannel(this.byteBufferIndex.array());
+ } else {
+ appendResult =
mappedFile.appendMessage(this.byteBufferIndex.array());
+ }
+ return appendResult;
+ } finally {
+ mappedFile.release();
}
- this.setMaxPhysicOffset(offset + size);
- boolean appendResult;
- if
(messageStore.getMessageStoreConfig().isPutConsumeQueueDataByFileChannel()) {
- appendResult =
mappedFile.appendMessageUsingFileChannel(this.byteBufferIndex.array());
- } else {
- appendResult =
mappedFile.appendMessage(this.byteBufferIndex.array());
- }
- return appendResult;
}
return false;
}
@@ -1271,7 +1279,19 @@ public class ConsumeQueue implements
ConsumeQueueInterface {
// transientStorePool is null, only need set wrote position here
MappedFile mappedFile = mappedFileQueue.getLastMappedFile(offset *
ConsumeQueue.CQ_STORE_UNIT_SIZE, true);
- fillPreBlank(mappedFile, offset * ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ if (mappedFile == null) {
+ log.error("initializeWithOffset failed: mappedFile is null for
offset {}", offset);
+ return;
+ }
+ if (!mappedFile.hold()) {
+ log.error("initializeWithOffset failed: mappedFile hold() failed
for offset {}", offset);
+ return;
+ }
+ try {
+ fillPreBlank(mappedFile, offset * ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ } finally {
+ mappedFile.release();
+ }
flush(0);
}
