iloveeyjafjalla opened a new issue, #11412: URL: https://github.com/apache/rocketmq/issues/11412
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment OS: any (broker hosting a topic whose `queue.type` attribute is `BatchCQ`) ### RocketMQ version branch: develop Git commit id: a7edeecfb ### JDK Version Compiler: Temurin 1.8.0_504 (target 1.8, same as CI) ### Describe the Bug A batch consume queue that has no mapped file yet cannot answer the three questions every other consume-queue implementation answers for an empty queue; it throws instead: - `BatchConsumeQueue.getEarliestUnit()` → `get(minOffsetInQueue)` (`store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java:364-366`) - `BatchConsumeQueue.getLatestUnit()` → `get(maxOffsetInQueue - 1)` (`:369-371`) - `BatchConsumeQueue.getLastOffset()` → dereferences the result of `getLatestUnit()` (`:374-377`) A freshly created `BatchConsumeQueue` has `minOffsetInQueue = -1` and `maxOffsetInQueue = 0` (`reviseMinOffsetInQueue()`, `:290-291`), so both unit accessors call `get(-1)`, which goes through `getBatchMsgIndexBuffer(-1)` to ```java protected BatchOffsetIndex getMinMsgOffset(MappedFile mappedFile, boolean getBatchSize, boolean getStoreTime) { if (mappedFile.getReadPosition() < CQ_STORE_UNIT_SIZE) { // :635-636, no null check ``` and `mappedFileQueue.getLastMappedFile()` returns `null` when the queue has no file (`MappedFileQueue.java:403`) — the queue object itself is created (and cached) by `ConsumeQueueStore.findConsumeQueue()` as soon as anything touches the topic, before a single message has been dispatched, so the empty state is the normal state of a new batch topic. Note that the sibling method in the same class already guards this: `getMaxMsgOffset` checks `mappedFile == null || …` (`:656-657`), so the null file is expected by the author; only the `getMinMsgOffset` path dereferences it. **Impact 1 — the admin command `cleanExpiredConsumerQueue` fails on the whole broker.** `ConsumeQueueStore.cleanExpired(minCommitLogOffset)` walks every consume queue of every non-system topic and calls `getLastOffset()` (`store/.../queue/ConsumeQueueStore.java:593`); the caller explicitly expects an empty queue to answer `-1` and has a dedicated branch for it: ```java if (maxCLOffsetInConsumeQueue == -1) { log.warn("maybe ConsumeQueue was created just now..."); } else if (maxCLOffsetInConsumeQueue < minCommitLogOffset) { … } ``` `BatchConsumeQueue.getLastOffset()` throws `NullPointerException` instead, so `mqadmin cleanExpiredConsumerQueue` returns `SYSTEM_ERROR` and cleans nothing, for as long as any batch topic on that broker has an empty queue. **Impact 2 — clients cannot read the earliest store time of such a queue.** `GET_EARLIEST_MSG_STORETIME` → `DefaultMessageStore.getEarliestMessageTime(topic, queueId)` → `logicQueue.getEarliestUnitAndStoreTime()` (`DefaultMessageStore.java:1295-1305`) throws instead of the documented `-1`; the same path backs `mqadmin queryConsumeTimeSpan`. **Contract violated** (siblings answer `null` / `-1`, and callers rely on it): - `ConsumeQueue.getLastOffset()` returns `-1` when there is no mapped file (`store/.../ConsumeQueue.java:501-511`), and `ConsumeQueue.getEarliestUnit()` returns `null` for an empty queue (`:976-984`). - `RocksDBConsumeQueue.getLatestUnit()` guards the offset (`maxOffset > 0 ? maxOffset - 1 : maxOffset`) and `getLastOffset()` returns `getMaxPhysicOffset()` (`store/.../queue/RocksDBConsumeQueue.java:362-377`). - `ConsumeQueueInterface` declares the units as nullable results (`store/.../queue/ConsumeQueueInterface.java:82-88`). ### Steps to Reproduce 1. Create a batch topic on a broker: `mqadmin updateTopic -t batchTopic -c DefaultCluster -a +queue.type=BatchCQ` (the attribute is `TopicAttributes.QUEUE_TYPE_ATTRIBUTE`, `common/.../TopicAttributes.java:30-35`). 2. Touch one of its queues so the queue object is created, e.g. `mqadmin cleanExpiredConsumerQueue -b 127.0.0.1:10911` or query it from a client; no message has to be sent. 3. The command fails with `SYSTEM_ERROR` produced by `java.lang.NullPointerException` at `BatchConsumeQueue.getMinMsgOffset`, and `earliestMsgStoreTime` for that queue throws as well. At the unit level the same state is reproduced with the existing harness in `store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeQueueTest.java`: `createBatchConsume(null)` + `load()` with no entries, then `getEarliestUnit()`, `getLatestUnit()` and `getLastOffset()` — all three throw today, while the equivalent `ConsumeQueue`/`RocksDBConsumeQueue` calls return `null`, `null` and `-1`. ### What Did You Expect to See? An empty batch consume queue reports “nothing in the queue”, exactly like its sibling implementations: `getEarliestUnit()`/`getLatestUnit()` return `null` (and `getEarliestUnitAndStoreTime()` therefore yields a `-1` store time, which `getEarliestMessageTime` already handles), and `getLastOffset()` returns `-1` so `cleanExpired` takes its documented “created just now” branch and keeps cleaning the other queues. ### What Did You See Instead? `NullPointerException` from `BatchConsumeQueue.getMinMsgOffset`, surfacing as `SYSTEM_ERROR` for `cleanExpiredConsumerQueue` (the whole command aborts, so no consume queue on that broker gets cleaned) and as a failing `earliestMsgStoreTime` / `queryConsumeTimeSpan` call. ### Additional Context Minimal fix, mirroring the sibling implementations and the existing `getMaxMsgOffset` guard: 1. `getMinMsgOffset`: treat a null mapped file as “no data” (`if (mappedFile == null || mappedFile.getReadPosition() < CQ_STORE_UNIT_SIZE)`), like `getMaxMsgOffset` already does; 2. `getEarliestUnit()` / `getLatestUnit()`: return `null` when the queue has no offset to read (`minOffsetInQueue >= 0` / `maxOffsetInQueue > 0`), the way `RocksDBConsumeQueue` does; 3. `getLastOffset()`: return `-1` when there is no latest unit, the way `ConsumeQueue` does. I will open a PR with these three changes plus a regression test in `BatchConsumeQueueTest` (red before: `NullPointerException`; green after: `null`/`null`/`-1`). -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
