You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2020/06/09 16:05:40 UTC
[GitHub] [rocketmq] lebron374 opened a new issue #2081: Question about rocketmq CommitLogDispatcherBuildConsumeQueue
lebron374 opened a new issue #2081:
URL: https://github.com/apache/rocketmq/issues/2081
i am confused about build consumeQueue procedure。
if put to consumeQueue fail，we just discard the message？
because without consumeQueue，message can't be consume。
```
public void putMessagePositionInfoWrapper(DispatchRequest request) {
final int maxRetries = 30;
boolean canWrite = this.defaultMessageStore.getRunningFlags().isCQWriteable();
for (int i = 0; i < maxRetries && canWrite; i++) {
long tagsCode = request.getTagsCode();
if (isExtWriteEnable()) {
ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
cqExtUnit.setFilterBitMap(request.getBitMap());
cqExtUnit.setMsgStoreTime(request.getStoreTimestamp());
cqExtUnit.setTagsCode(request.getTagsCode());
long extAddr = this.consumeQueueExt.put(cqExtUnit);
if (isExtAddr(extAddr)) {
tagsCode = extAddr;
} else {
log.warn("Save consume queue extend fail, So just save tagsCode! {}, topic:{}, queueId:{}, offset:{}", cqExtUnit,
topic, queueId, request.getCommitLogOffset());
}
}
boolean result = this.putMessagePositionInfo(request.getCommitLogOffset(),
request.getMsgSize(), tagsCode, request.getConsumeQueueOffset());
if (result) {
if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE ||
this.defaultMessageStore.getMessageStoreConfig().isEnableDLegerCommitLog()) {
this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(request.getStoreTimestamp());
}
this.defaultMessageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
return;
} else {
// XXX: warn and notify me
log.warn("[BUG]put commit log position info to " + topic + ":" + queueId + " " + request.getCommitLogOffset()
+ " failed, retry " + i + " times");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
log.warn("", e);
}
}
}
// XXX: warn and notify me
log.error("[BUG]consume queue can not write, {} {}", this.topic, this.queueId);
this.defaultMessageStore.getRunningFlags().makeLogicsQueueError();
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] imaffe commented on issue #2081: Question about rocketmq CommitLogDispatcherBuildConsumeQueue
Posted by GitBox <gi...@apache.org>.
imaffe commented on issue #2081:
URL: https://github.com/apache/rocketmq/issues/2081#issuecomment-645400346
@lebron374 Hope this answer helps.
Issues are generally used for bug reports and feature requests, so if you have any further questions about Apache RocketMQ, it is recommended to discuss it on the mailing list first, the mailing list [address] (https://rocketmq.apache.org/about/contact). I will close the Issue first, but please feel free to reopen it if you have any other issues.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] imaffe edited a comment on issue #2081: Question about rocketmq CommitLogDispatcherBuildConsumeQueue
Posted by GitBox <gi...@apache.org>.
imaffe edited a comment on issue #2081:
URL: https://github.com/apache/rocketmq/issues/2081#issuecomment-645400346
@lebron374 Hope this answer was helpful ~ But I have to close this issue as it has not received response in 5 days.
Issues are generally used for bug reports and feature requests, so if you have any further questions about Apache RocketMQ, it is recommended to discuss it on the mailing list first, the mailing list [address] (https://rocketmq.apache.org/about/contact). I will close the Issue first, but please feel free to reopen it if you have any other issues.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhiboo commented on issue #2081: Question about rocketmq CommitLogDispatcherBuildConsumeQueue
Posted by GitBox <gi...@apache.org>.
lizhiboo commented on issue #2081:
URL: https://github.com/apache/rocketmq/issues/2081#issuecomment-643084747
> putMessagePositionInfoWrapper
Message stored in CommitLog, reputMessageService put message to ConsumeQueue from its maxPhysicalPosInLogicQueue at init stage. So it will not discard message, even if reputMessageService thread crash. reputMessageService thread catch mostly exception, it rarely carsh.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] RongtongJin closed issue #2081: Question about rocketmq CommitLogDispatcherBuildConsumeQueue
Posted by GitBox <gi...@apache.org>.
RongtongJin closed issue #2081:
URL: https://github.com/apache/rocketmq/issues/2081
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org