From 36adf1251ff5d3684083c2dbd2debf5ba2c104a8 Mon Sep 17 00:00:00 2001 From: guyinyou <36399867+guyinyou@users.noreply.github.com> Date: Tue, 20 Jan 2026 14:44:22 +0800 Subject: [PATCH] =?UTF-8?q?[ISSUE=20#10017]=20Validate=20commitlog=20offse?= =?UTF-8?q?t=20in=20recoverAbnormally=20to=20prevent=20processing=20?= =?UTF-8?q?=E2=80=A6=20(#10018)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * validate commitlog offset in recoverAbnormally to prevent processing old file data that passes CRC checks Change-Id: If4b1881f82d26ce8d374472d73ec9ce3d51ba643 * fix Change-Id: Idc4bf7ec476cc9b6529619c2aa9afd6a980b819c * add checkCommitLogOffsetOnRecover in MessageStoreConfig Change-Id: Iac9afbb8b3ffb03fa15890decaf502afbfa44cf9 --------- Co-authored-by: guyinyou --- .../org/apache/rocketmq/store/CommitLog.java | 18 +++++++++++++++--- .../store/config/MessageStoreConfig.java | 11 +++++++++++ 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java index 286f31cd4a..3b92f1a745 100644 --- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java +++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java @@ -738,6 +738,7 @@ public class CommitLog implements Swappable { // recover by the minimum time stamp boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover(); boolean checkDupInfo = this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable(); + boolean checkCommitLogOffsetOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCommitLogOffsetOnRecover(); int maxRecoverNum = this.defaultMessageStore.getMessageStoreConfig().getCommitLogRecoverMaxNum(); if (maxRecoverNum <= 0) { maxRecoverNum = 10; @@ -792,8 +793,18 @@ public class CommitLog implements Swappable { while (true) { DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover, checkDupInfo); int size = dispatchRequest.getMsgSize(); - if (dispatchRequest.isSuccess()) { + // Check commitlog offset validity if enabled + if (checkCommitLogOffsetOnRecover) { + if (dispatchRequest.getCommitLogOffset() < mappedFile.getFileFromOffset() + || dispatchRequest.getCommitLogOffset() > mappedFile.getFileFromOffset() + mappedFile.getFileSize()) { + log.warn("found illegal commitlog offset {} in {}, current pos={}, it will be truncated.", + dispatchRequest.getCommitLogOffset(), mappedFile.getFileName(), processOffset + mappedFileOffset); + log.info("recover physics file end, {} pos={}", mappedFile.getFileName(), byteBuffer.position()); + + break; + } + } // Normal data if (size > 0) { lastValidMsgPhyOffset = processOffset + mappedFileOffset; @@ -925,7 +936,8 @@ public class CommitLog implements Swappable { return isMappedFileMatchedRecover(phyOffset, storeTimestamp, recoverNormally); } - private boolean isMappedFileMatchedRecover(long phyOffset, long storeTimestamp, boolean recoverNormally) throws RocksDBException { + private boolean isMappedFileMatchedRecover(long phyOffset, long storeTimestamp, + boolean recoverNormally) throws RocksDBException { boolean result = this.defaultMessageStore.getQueueStore().isMappedFileMatchedRecover(phyOffset, storeTimestamp, recoverNormally); if (null != this.defaultMessageStore.getTransMessageRocksDBStore() && defaultMessageStore.getMessageStoreConfig().isTransRocksDBEnable() && !defaultMessageStore.getMessageStoreConfig().isTransWriteOriginTransHalfEnable()) { result = result && this.defaultMessageStore.getTransMessageRocksDBStore().isMappedFileMatchedRecover(phyOffset); @@ -2386,7 +2398,7 @@ public class CommitLog implements Swappable { long costTime = this.systemClock.now() - beginClockTimestamp; log.info("[{}] scanFilesInPageCache-cost {} ms.", costTime > 30 * 1000 ? "NOTIFYME" : "OK", costTime); } catch (Throwable e) { - log.warn("{} service has e: ", this.getServiceName() , e); + log.warn("{} service has e: ", this.getServiceName(), e); } } log.info("{} service end", this.getServiceName()); diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java index 65dba5390d..d7f17efd64 100644 --- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java +++ b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java @@ -194,6 +194,9 @@ public class MessageStoreConfig { // This ensures no on-the-wire or on-disk corruption to the messages occurred. // This check adds some overhead,so it may be disabled in cases seeking extreme performance. private boolean checkCRCOnRecover = true; + // Whether check the commitlog offset validity during abnormal recovery. + // This helps detect and truncate old file data that may pass CRC checks but contains invalid offsets. + private boolean checkCommitLogOffsetOnRecover = false; // How many pages are to be flushed when flush CommitLog private int flushCommitLogLeastPages = 4; // How many pages are to be committed when commit data to file @@ -796,6 +799,14 @@ public class MessageStoreConfig { this.checkCRCOnRecover = checkCRCOnRecover; } + public boolean isCheckCommitLogOffsetOnRecover() { + return checkCommitLogOffsetOnRecover; + } + + public void setCheckCommitLogOffsetOnRecover(boolean checkCommitLogOffsetOnRecover) { + this.checkCommitLogOffsetOnRecover = checkCommitLogOffsetOnRecover; + } + public boolean isForceVerifyPropCRC() { return forceVerifyPropCRC; }