mirror of
https://github.com/apache/rocketmq.git
synced 2026-10-24 02:15:19 +08:00
[ISSUE #11412] fix(store): report an empty batch consume queue as empty instead of throwing (#11417)
This commit is contained in:
@@ -362,18 +362,18 @@ public class BatchConsumeQueue implements ConsumeQueueInterface {
|
||||
|
||||
@Override
|
||||
public CqUnit getEarliestUnit() {
|
||||
return get(minOffsetInQueue);
|
||||
return minOffsetInQueue >= 0 ? get(minOffsetInQueue) : null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CqUnit getLatestUnit() {
|
||||
return get(maxOffsetInQueue - 1);
|
||||
return maxOffsetInQueue > 0 ? get(maxOffsetInQueue - 1) : null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long getLastOffset() {
|
||||
CqUnit latestUnit = getLatestUnit();
|
||||
return latestUnit.getPos() + latestUnit.getSize();
|
||||
return latestUnit == null ? -1 : latestUnit.getPos() + latestUnit.getSize();
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -633,7 +633,7 @@ public class BatchConsumeQueue implements ConsumeQueueInterface {
|
||||
}
|
||||
|
||||
protected BatchOffsetIndex getMinMsgOffset(MappedFile mappedFile, boolean getBatchSize, boolean getStoreTime) {
|
||||
if (mappedFile.getReadPosition() < CQ_STORE_UNIT_SIZE) {
|
||||
if (mappedFile == null || mappedFile.getReadPosition() < CQ_STORE_UNIT_SIZE) {
|
||||
return null;
|
||||
}
|
||||
return getBatchOffsetIndexByPos(mappedFile, 0, getBatchSize, getStoreTime);
|
||||
|
||||
@@ -250,6 +250,18 @@ public class BatchConsumeQueueTest extends StoreTestBase {
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEmptyBatchConsumeQueueReportsNoUnitAndMinusOneLastOffset() {
|
||||
BatchConsumeQueue batchConsumeQueue = createBatchConsume(null);
|
||||
batchConsumeQueue.load();
|
||||
Assert.assertEquals(0, getBcqFileSize(batchConsumeQueue));
|
||||
|
||||
Assert.assertNull(batchConsumeQueue.getEarliestUnit());
|
||||
Assert.assertNull(batchConsumeQueue.getLatestUnit());
|
||||
Assert.assertEquals(-1, batchConsumeQueue.getLastOffset());
|
||||
Assert.assertEquals(-1, batchConsumeQueue.getEarliestUnitAndStoreTime().getObject2().longValue());
|
||||
}
|
||||
|
||||
@After
|
||||
@Override
|
||||
public void clear() {
|
||||
|
||||
Reference in New Issue
Block a user