You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "cserwen (via GitHub)" <gi...@apache.org> on 2023/09/13 12:26:06 UTC
[GitHub] [rocketmq] cserwen opened a new issue, #7355: [Bug] ConsumeQueue building exception caused by host downtime
cserwen opened a new issue, #7355:
URL: https://github.com/apache/rocketmq/issues/7355
### Before Creating the Bug Report
- [X] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions).
- [X] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate.
- [X] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.
### Runtime platform environment
OS: CentOS 7.3
### RocketMQ version
4.8.x
### JDK Version
openjdk version "1.8.0_202"
### Describe the Bug
- A downtime occurred on the slave node
- When the Broker process is started, the following exception log appears during building consumeQueue index.
```log
[BUG]logic queue order maybe wrong, expectLogicOffset: 5762122360 currentLogicOffset: 5762121440 Topic: hlth-center-data QID: 1 Diff: 920
```
- Delete all consumeQueues and back to normal.
### Steps to Reproduce
- stop the slave node.
- delete the last consumeQueue of a queue.
- start the slave node.
### What Did You Expect to See?
This log should not be printed.
### What Did You See Instead?
```log
[BUG]logic queue order maybe wrong, expectLogicOffset: 5762122360 currentLogicOffset: 5762121440 Topic: hlth-center-data QID: 1 Diff: 920
```
### Additional Context
### # Possible reason
- A power outage on the host causes the loss of unpersisted consumeQueue records in the pagecache.
### # Process
- ConsumeQueue files are flushed asynchronously, executed once every 1 second
- The index is built and stored in the memory, and the disk is flushed once every second, so there will be some index data (the white blocks) is not written to the disk.
- A power outage on the host causes the loss of unpersisted consumeQueue records in the pagecache.
- After the Broker is started, the current maximum reputOffset is 1004. The index building starts from 1005, so 1002 and 1003 are skipped.
### # Verify
- Read the wrong consumeQueue file,
```
last=287986464, new=287986511, file=/home/mi/bak_cq/hlth-center-data/1/00000000005754000000
MessageExt [brokerName=null, queueId=1, storeSize=1914, queueOffset=287986464, sysFlag=0, bornTimestamp=1694089617516}]
MessageExt [brokerName=null, queueId=1, storeSize=1218, queueOffset=287986511, sysFlag=0, bornTimestamp=1694089620422}]
```
- There is a gap. Theoretically, the QueueOffset of the next message should be 287986465, but in fact it is 287986511, a difference of 46. The converted index file length is exactly 920 bytes, which is consistent with the log.
### # Hot to fix
#### 1. Memory cache
- Add a cache to maintains the flushed offset of each queeu: map <Topic#QueueId, commitLogOffsetPhy>
- When exiting abnormally, enter the CQ recover logic and start scanning the commitLog for recovery based on the smallest persistent location recorded in the map.
#### 2. Scan partly commitLogs
- Since cq will be flushed periodically, the most recent commitLog files will be affected.
- Find the corresponding CommitLog A according to the error log. Then push forward a certain amount of time (3 minutes), find the first CommitLog that exceeds 3 minutes, and then start building from this commitLog
### # Linked issue
#1397
--
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-unsubscribe@rocketmq.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733051875
> Why not recover from the minimum cmq physical offset when abnormally exiting🤔?
collect all consumer queue last message's commit log physical offset, and use the minimum as the reputService's reputFromOffset(now is the maxPhysicalPosInLogicQueue).
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1734968462
> > Hello,I have a question. If System is exit abnormal, when restart, rocketMq will call recoverAbnormally() to recover consumerQueue and index data. So why we need to fix the data? 🧐 Maybe I need to try reproduction this bug. That's the code :
>
> It seems that this case has indeed been considered, but we are using DLedger mode, and the recovery logic of dLedgerCommitLog does not have this process. This should be the cause.
That should be it, I made some time to research it 🥲... Thanks, I will be to learn about DLedger.
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] cserwen commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "cserwen (via GitHub)" <gi...@apache.org>.
cserwen commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1734962052
> Hello,I have a question. If System is exit abnormal, when restart, rocketMq will call recoverAbnormally() to recover consumerQueue and index data. So why we need to fix the data? 🧐 Maybe I need to try reproduction this bug. That's the code :
It seems that this case has indeed been considered, but we are using DLedger mode, and the recovery logic of dLedgerCommitLog does not have this process. This should be the cause.
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] cserwen commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "cserwen (via GitHub)" <gi...@apache.org>.
cserwen commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733118782
> > > > Why not recover from the minimum cmq physical offset when abnormally exiting🤔?
> > >
> > >
> > > collect all consumer queue last message's commit log physical offset, and use the minimum as the reputService's reputFromOffset(now is the maxPhysicalPosInLogicQueue).
> >
> >
> > @suopovate Emmm, looks like this is the first way I mentioned.
> > Maybe, I think my way is the simplest, just change the 'reputFromOffset' value, finally the data will be normal, and no data loss.🤓
> > We needn't not keep the cache of consumer queue flushed offset.
@suopovate
- In fact, this also has some disadvantages. Some queues may only generate a few messages, which means that its endOffset is very small. In this case, we will rebuilding almost all cqs.
- By the cache, we can add the dimension of time to more reasonably determine the offset which the broker should begin to recover at.
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] ConsumeQueue building exception caused by host downtime [rocketmq]
Posted by "yuz10 (via GitHub)" <gi...@apache.org>.
yuz10 closed issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
URL: https://github.com/apache/rocketmq/issues/7355
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733959552
Hello,I have a question.
If System is exit abnormal, when restart, rocketMq will call recoverAbnormally() to recover consumerQueue and index data.
So why we need to fix the data? 🧐
That's the code :
@Deprecated
public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
// recover by the minimum time stamp
boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
boolean checkDupInfo = this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable();
final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
if (!mappedFiles.isEmpty()) {
// Looking beginning to recover from which file
int index = mappedFiles.size() - 1;
MappedFile mappedFile = null;
for (; index >= 0; index--) {
mappedFile = mappedFiles.get(index);
if (this.isMappedFileMatchedRecover(mappedFile)) {
log.info("recover from this mapped file " + mappedFile.getFileName());
break;
}
}
if (index < 0) {
index = 0;
mappedFile = mappedFiles.get(index);
}
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
long processOffset = mappedFile.getFileFromOffset();
long mappedFileOffset = 0;
long lastValidMsgPhyOffset = processOffset;
long lastConfirmValidMsgPhyOffset = processOffset;
// abnormal recover require dispatching
boolean doDispatch = true;
while (true) {
DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover, checkDupInfo);
int size = dispatchRequest.getMsgSize();
if (dispatchRequest.isSuccess()) {
// Normal data
if (size > 0) {
lastValidMsgPhyOffset = processOffset + mappedFileOffset;
mappedFileOffset += size;
if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable() || this.defaultMessageStore.getBrokerConfig().isEnableControllerMode()) {
if (dispatchRequest.getCommitLogOffset() + size <= this.defaultMessageStore.getCommitLog().getConfirmOffset()) {
this.getMessageStore().onCommitLogDispatch(dispatchRequest, doDispatch, mappedFile, true, false);
lastConfirmValidMsgPhyOffset = dispatchRequest.getCommitLogOffset() + size;
}
} else {
// it is the place that rebuild consumerQueue and index data...🤔
this.getMessageStore().onCommitLogDispatch(dispatchRequest, doDispatch, mappedFile, true, false);
}
}
// Come the end of the file, switch to the next file
// Since the return 0 representatives met last hole, this can
// not be included in truncate offset
else if (size == 0) {
this.getMessageStore().onCommitLogDispatch(dispatchRequest, doDispatch, mappedFile, true, true);
index++;
if (index >= mappedFiles.size()) {
// The current branch under normal circumstances should
// not happen
log.info("recover physics file over, last mapped file " + mappedFile.getFileName());
break;
} else {
mappedFile = mappedFiles.get(index);
byteBuffer = mappedFile.sliceByteBuffer();
processOffset = mappedFile.getFileFromOffset();
mappedFileOffset = 0;
log.info("recover next physics file, " + mappedFile.getFileName());
}
}
} else {
if (size > 0) {
log.warn("found a half message at {}, it will be truncated.", processOffset + mappedFileOffset);
}
log.info("recover physics file end, " + mappedFile.getFileName() + " pos=" + byteBuffer.position());
break;
}
}
processOffset += mappedFileOffset;
if (this.defaultMessageStore.getBrokerConfig().isEnableControllerMode()) {
if (this.defaultMessageStore.getConfirmOffset() < this.defaultMessageStore.getMinPhyOffset()) {
log.error("confirmOffset {} is less than minPhyOffset {}, correct confirmOffset to minPhyOffset", this.defaultMessageStore.getConfirmOffset(), this.defaultMessageStore.getMinPhyOffset());
this.defaultMessageStore.setConfirmOffset(this.defaultMessageStore.getMinPhyOffset());
} else if (this.defaultMessageStore.getConfirmOffset() > lastConfirmValidMsgPhyOffset) {
log.error("confirmOffset {} is larger than lastConfirmValidMsgPhyOffset {}, correct confirmOffset to lastConfirmValidMsgPhyOffset", this.defaultMessageStore.getConfirmOffset(), lastConfirmValidMsgPhyOffset);
this.defaultMessageStore.setConfirmOffset(lastConfirmValidMsgPhyOffset);
}
} else {
this.setConfirmOffset(lastValidMsgPhyOffset);
}
this.mappedFileQueue.setFlushedWhere(processOffset);
this.mappedFileQueue.setCommittedWhere(processOffset);
this.mappedFileQueue.truncateDirtyFiles(processOffset);
// Clear ConsumeQueue redundant data
if (maxPhyOffsetOfConsumeQueue >= processOffset) {
log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
}
}
// Commitlog case files are deleted
else {
log.warn("The commitlog files are deleted, and delete the consume queue files");
this.mappedFileQueue.setFlushedWhere(0);
this.mappedFileQueue.setCommittedWhere(0);
this.defaultMessageStore.destroyLogics();
}
}
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1731078909
Cool, seems like I found this bug when I read the source code, but I don't go into it. 🙉
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733046350
Why not recover from the minimum cmq physical offset when abnormally exiting🤔?
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733138240
> > > > > Why not recover from the minimum cmq physical offset when abnormally exiting🤔?
> > > >
> > > >
> > > > collect all consumer queue last message's commit log physical offset, and use the minimum as the reputService's reputFromOffset(now is the maxPhysicalPosInLogicQueue).
> > >
> > >
> > > @suopovate Emmm, looks like this is the first way I mentioned.
> > > Maybe, I think my way is the simplest, just change the 'reputFromOffset' value, finally the data will be normal, and no data loss.🤓
> > > We needn't not keep the cache of consumer queue flushed offset.
>
> @suopovate
>
> * In fact, this also has some disadvantages. Some queues may only generate a few messages, which means that its endOffset is very small. In this case, we will rebuilding almost all cqs.
> * By the cache, we can add the dimension of time to more reasonably determine the offset which the broker should begin to recover at.
Emmm, indeed I left out the question.
I will try to fix it in local, thanks for your suggestion.🫡
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] cserwen commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "cserwen (via GitHub)" <gi...@apache.org>.
cserwen commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733072839
> > Why not recover from the minimum cmq physical offset when abnormally exiting🤔?
>
> collect all consumer queue last message's commit log physical offset, and use the minimum as the reputService's reputFromOffset(now is the maxPhysicalPosInLogicQueue).
@suopovate Starting to recover directly from the minimum physical address of commitLog isn't different from deleting and rebuilding cq. It takes too long and does not seem to be an elegant way.
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1731079166
Cool, seems like I found this bug when I read the source code, but I don't go into it. 🙉
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] suopovate commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "suopovate (via GitHub)" <gi...@apache.org>.
suopovate commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1733098334
> > > Why not recover from the minimum cmq physical offset when abnormally exiting🤔?
> >
> >
> > collect all consumer queue last message's commit log physical offset, and use the minimum as the reputService's reputFromOffset(now is the maxPhysicalPosInLogicQueue).
>
> @suopovate Emmm, looks like this is the first way I mentioned.
Maybe, I think my way is the simplest, just change the 'reputFromOffset' value, finally the data will be normal, and no data loss.🤓
We needn't not keep the cache of consumer queue flushed offset.
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] bxfjb commented on issue #7355: [Bug] ConsumeQueue building exception caused by host downtime
Posted by "bxfjb (via GitHub)" <gi...@apache.org>.
bxfjb commented on issue #7355:
URL: https://github.com/apache/rocketmq/issues/7355#issuecomment-1717561185
seems clear, is it okay to hand it over to me?
--
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-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org