There may be a bug in DefaultMessageStore#getMessage

[Knowden]

maxFilterMessageCount is calculated as maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE and it's seems like meaning the maxFilterMessageSize.

final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE);

and it's used to judge wether this pull has scanned enough message as follow

if (cqUnit.getQueueOffset() - offset > maxFilterMessageCount) {
    break;
}

the cqUnit.queueOffset is logic offset of consumer queue which is calculated as follow

long queueOffset = (sbr.getStartOffset() + sbr.getByteBuffer().position() -  relativePos) / CQ_STORE_UNIT_SIZE;
CqUnit cqUnit = new CqUnit(queueOffset,
        sbr.getByteBuffer().getLong(),
        sbr.getByteBuffer().getInt(),
        sbr.getByteBuffer().getLong());

Is there something going wrong? or if I made some mistake when reading source code and submitting issue, please figure it out thanks

[francisoliverlee]

For some pulled messages for maxMsgNums could be filter-out. so Message Count is not a good way to limit. Change Count-Limit to Bytes-Limit, you can check code on 4.9.x.

rocketmq/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

Line 623 in c88686f

for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {

but it seems need some updates on

if ((cqUnit.getQueueOffset() - offset) *  CQ_STORE_UNIT_SIZE  > maxFilterMessageCount) {
    break;
}

[HScarb]

IMO you're right.
maxFilterMessageCount is physical offset of ConsumeQueue file, while cqUnit.getQueueOffset() - offset is logical offset

[Knowden]

For some pulled messages for maxMsgNums could be filter-out. so Message Count is not a good way to limit. Change Count-Limit to Bytes-Limit, you can check code on 4.9.x.

rocketmq/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

Line 623 in c88686f

for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {

but it seems need some updates on

if ((cqUnit.getQueueOffset() - offset) *  CQ_STORE_UNIT_SIZE  > maxFilterMessageCount) {
    break;
}

ok. I will submit a PR to fix this later on.

[Knowden]

#5171

[ni-ze]

LGTM~