ymwneu opened a new issue, #10959: URL: https://github.com/apache/rocketmq/issues/10959
### 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 CentOS ### RocketMQ version 5.5.0 ### JDK Version JDK11 ### Describe the Bug `RocksDBConsumeQueue#iterateFrom(long startIndex)` / `iterateFrom(long startIndex, int count)` (store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueue.java, around line 299-317) only validate `startIndex >= 0 && startIndex < maxOffsetInQueue`. They never check `startIndex` against the queue's current `minOffsetInQueue`. This is inconsistent with the file-based `ConsumeQueue#iterateFrom` (store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java, `getIndexBuffer`, around line 918-935), which explicitly checks `offset >= this.getMinLogicOffset()` and returns `null` otherwise. Because of this gap, when `startIndex` is below the RocksDB consume queue's current min offset (e.g. after the underlying data for that offset has been cleaned/expired, or the queue's min offset was advanced by recovery/rebuild), `iterateFrom` still returns a non-null `LargeRocksDBConsumeQueueIterator`. That iterator's `hasNext()` only compares `currentIndex < totalCount` (pure index arithmetic) and never verifies the record actually exists, while `next()` calls `consumeQueueStore.get(...)`, which returns `null` (or an undersized buffer) for a purged offset and makes `next()` return `null` — violating the normal Iterator contract of "hasNext()==true implies next() is non-null". `ScheduleMessageService.DeliverDelayedMessageTimerTask#executeOnTimeUp` (broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java, around line 410-431) relies on `iterateFrom` returning `null` to detect and self-correct an out-of-range offset (see the `if (bufferCQ == null) {...}` block at line 411-424). Since it receives a non-null iterator instead, execution falls through to the `while` loop, calls `bufferCQ.next()` (line 430) which returns `null`, and then immediately dereferences it via `cqUnit.getPos()` (line 431), throwing a `NullPointerException`. Worse, `DeliverDelayedMessageTimerTask#run()` (line 375-386) catches this `Throwable`, logs it, and reschedules the *same* task with the identical (still out-of-range) `this.offset` via `scheduleNextTimerTask(this.offset, DELAY_FOR_A_PERIOD)`, since `this.offset` is a `final` field that was never advanced before the exception. This creates an infinite NPE retry loop for that delay level: the timer task fires every `DELAY_FOR_A_PERIOD`, throws the same NPE, logs an error, and reschedules itself with the same broken offset — effectively stalling delayed/scheduled message delivery for that delay level until the broker is restarted (and the underlying condition is otherwise resolved). ### Steps to Reproduce 1. Run a broker configured with the RocksDB consume queue store (`messageStoreConfig.setStoreType(StoreType.DEFAULT_ROCKSDB)`). 2. Minimal unit-level reproduction (no full cluster needed): ```java RocksDBConsumeQueueStore store = mock(RocksDBConsumeQueueStore.class); when(store.getMinOffsetInQueue(anyString(), anyInt())).thenReturn(9000L); when(store.getMaxOffsetInQueue(anyString(), anyInt())).thenReturn(10000L); // store.get(topic, queueId, offset) for any offset < 9000 returns null, // simulating a purged/expired record. RocksDBConsumeQueue cq = new RocksDBConsumeQueue(storeConfig, store, "topic", 0); ReferredIterator<CqUnit> it = cq.iterateFrom(8000); // 8000 < minOffset(9000) // Bug: `it` is non-null. assertTrue(it.hasNext()); // true, based purely on index arithmetic assertNull(it.next()); // null CqUnit, because offset 8000 is already purged ``` 3. Real-world trigger path: `ScheduleMessageService` persists a resume offset per delay level (config/delayOffset.json). If the schedule topic's RocksDB consume queue min offset later advances past that persisted offset (e.g. via expired-data cleanup, or the consume queue being rebuilt/recovered with a higher min offset than the still-recorded schedule offset), the next scheduled tick calls `cq.iterateFrom(this.offset)` with `this.offset < cq.getMinOffsetInQueue()`. 4. Observe the broker log repeatedly print: ``` ScheduleMessageService, executeOnTimeUp exception java.lang.NullPointerException at org.apache.rocketmq.broker.schedule.ScheduleMessageService$DeliverDelayedMessageTimerTask.executeOnTimeUp(ScheduleMessageService.java:431) ``` ...on a fixed interval (DELAY_FOR_A_PERIOD), never recovering on its own. ### What Did You Expect to See? `RocksDBConsumeQueue#iterateFrom` should return `null` when `startIndex` is below the queue's current `getMinOffsetInQueue()`, exactly like the file-based `ConsumeQueue` implementation does via its `getMinLogicOffset()` check. With that, `ScheduleMessageService.executeOnTimeUp()` would take its existing `if (bufferCQ == null)` branch (line 411-424), log a single "schedule CQ offset invalid" error, self-correct the offset (clamped to the queue's current min/max bound), and keep delivering delayed messages normally on the next tick — no exception, no stuck retry loop. ### What Did You See Instead? `iterateFrom` returns a non-null iterator whose `next()` yields `null` once the requested offset has already been purged from RocksDB. `ScheduleMessageService` dereferences that null `CqUnit` at line 431 and throws a `NullPointerException`: ``` java.lang.NullPointerException at org.apache.rocketmq.broker.schedule.ScheduleMessageService$DeliverDelayedMessageTimerTask.executeOnTimeUp(ScheduleMessageService.java:431) at org.apache.rocketmq.broker.schedule.ScheduleMessageService$DeliverDelayedMessageTimerTask.run(ScheduleMessageService.java:379) ... ``` Because `DeliverDelayedMessageTimerTask#run()` catches this exception and reschedules itself with the same unchanged offset, the same NPE repeats indefinitely every `DELAY_FOR_A_PERIOD`, and delayed/scheduled message delivery for the affected delay level is effectively stuck until the broker is restarted. ### Additional Context _No response_ -- 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]
