(rocketmq) branch develop updated: [ISSUE #11412] fix(store): report an empty batch consume queue as empty instead of throwing (#11417)

[jinrongtong]

This is an automated email from the ASF dual-hosted git repository.

RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 5d68880cbd [ISSUE #11412] fix(store): report an empty batch consume 
queue as empty instead of throwing (#11417)
5d68880cbd is described below

commit 5d68880cbdfa5c6d87b5681c5eb3684df0e18ef8
Author: bai mengxuan <2013373...@qq.com>
AuthorDate: Fri Oct 16 14:12:40 2026 +0800

    [ISSUE #11412] fix(store): report an empty batch consume queue as empty 
instead of throwing (#11417)
---
 .../org/apache/rocketmq/store/queue/BatchConsumeQueue.java   |  8 ++++----
 .../apache/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() {