lollipopjin commented on code in PR #9256:
URL: https://github.com/apache/rocketmq/pull/9256#discussion_r2053463902
##########
store/src/main/java/org/apache/rocketmq/store/CommitLog.java:
##########
@@ -842,49 +833,32 @@ protected void onCommitLogAppend(MessageExtBrokerInner
msg, AppendMessageResult
this.getMessageStore().onCommitLogAppend(msg, result, commitLogFile);
}
- private boolean isMappedFileMatchedRecover(final MappedFile mappedFile)
throws RocksDBException {
+ private boolean isMappedFileMatchedRecover(final MappedFile mappedFile,
+ boolean recoverNormally) throws RocksDBException {
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
int magicCode =
byteBuffer.getInt(MessageDecoder.MESSAGE_MAGIC_CODE_POSITION);
if (magicCode != MessageDecoder.MESSAGE_MAGIC_CODE && magicCode !=
MessageDecoder.MESSAGE_MAGIC_CODE_V2) {
return false;
}
- if
(this.defaultMessageStore.getMessageStoreConfig().isEnableRocksDBStore()) {
- final long maxPhyOffsetInConsumeQueue =
this.defaultMessageStore.getQueueStore().getMaxPhyOffsetInConsumeQueue();
- long phyOffset =
byteBuffer.getLong(MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION);
- if (phyOffset <= maxPhyOffsetInConsumeQueue) {
- log.info("find check. beginPhyOffset: {},
maxPhyOffsetInConsumeQueue: {}", phyOffset, maxPhyOffsetInConsumeQueue);
- return true;
- }
- } else {
- int sysFlag = byteBuffer.getInt(MessageDecoder.SYSFLAG_POSITION);
- int bornHostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG)
== 0 ? 8 : 20;
- int msgStoreTimePos = 4 + 4 + 4 + 4 + 4 + 8 + 8 + 4 + 8 +
bornHostLength;
- long storeTimestamp = byteBuffer.getLong(msgStoreTimePos);
- if (0 == storeTimestamp) {
- return false;
- }
+ int sysFlag = byteBuffer.getInt(MessageDecoder.SYSFLAG_POSITION);
+ int bornHostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0
? 8 : 20;
+ int msgStoreTimePos = 4 + 4 + 4 + 4 + 4 + 8 + 8 + 4 + 8 +
bornHostLength;
+ long storeTimestamp = byteBuffer.getLong(msgStoreTimePos);
+ if (0 == storeTimestamp) {
+ return false;
+ }
+ long phyOffset =
byteBuffer.getLong(MessageDecoder.MESSAGE_PHYSIC_OFFSET_POSITION);
- if
(this.defaultMessageStore.getMessageStoreConfig().isMessageIndexEnable()
- &&
this.defaultMessageStore.getMessageStoreConfig().isMessageIndexSafe()) {
- if (storeTimestamp <=
this.defaultMessageStore.getStoreCheckpoint().getMinTimestampIndex()) {
- log.info("find check timestamp, {} {}",
- storeTimestamp,
- UtilAll.timeMillisToHumanString(storeTimestamp));
- return true;
- }
- } else {
- if (storeTimestamp <=
this.defaultMessageStore.getStoreCheckpoint().getMinTimestamp()) {
- log.info("find check timestamp, {} {}",
- storeTimestamp,
- UtilAll.timeMillisToHumanString(storeTimestamp));
- return true;
- }
- }
+ if
(this.defaultMessageStore.getMessageStoreConfig().isMessageIndexEnable() &&
+
this.defaultMessageStore.getMessageStoreConfig().isMessageIndexSafe() &&
+ storeTimestamp >
this.defaultMessageStore.getStoreCheckpoint().getIndexMsgTimestamp()) {
Review Comment:
Add some log here for detailed information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
