From 5d68880cbdfa5c6d87b5681c5eb3684df0e18ef8 Mon Sep 17 00:00:00 2001 From: bai mengxuan <2013373049@qq.com> Date: Fri, 16 Oct 2026 14:12:40 +0800 Subject: [PATCH] [ISSUE #11412] fix(store): report an empty batch consume queue as empty instead of throwing (#11417) --- .../rocketmq/store/queue/BatchConsumeQueue.java | 8 ++++---- .../rocketmq/store/queue/BatchConsumeQueueTest.java | 12 ++++++++++++ 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java index eeab1fc194..071f0ad6f7 100644 --- a/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java +++ b/store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java @@ -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); diff --git a/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java b/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java index c6525bd836..94ee625bcf 100644 --- a/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java @@ -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() {