tju-yxq opened a new issue, #1353: URL: https://github.com/apache/rocketmq-dashboard/issues/1353
## Bug Report ### 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: Ubuntu 20.04 / Any OS running RocketMQ Studio ### RocketMQ version branch: rocketmq-studio version: 5.3.2+ Git commit id: f727341 ### JDK Version OpenJDK 21 ### Describe the Bug Both `RocketMQMessageProvider.queryByTopic()` and `RocketMQDLQProvider.collectDeadLetters()` contain a shared defect in their pull-consumer scan loops: when `PullResult.getPullStatus()` returns `OFFSET_ILLEGAL`, the loop **breaks** out of the current queue instead of **continuing** with the corrected offset from `nextBeginOffset`. This causes messages to be silently dropped from query results and dead-letter resend collections. The problematic pattern (identical in both methods): ```java long nextOffset = pullResult.getNextBeginOffset(); if (nextOffset <= offset) { break; // (covered by #1159 - offset did not advance) } offset = nextOffset; if (pullResult.getPullStatus() != PullStatus.FOUND || pullResult.getMsgFoundList() == null) { break; // BUG: breaks on ALL non-FOUND statuses, including OFFSET_ILLEGAL } ``` RocketMQ's `PullStatus` has four values: | Status | Meaning | Correct action | |--------|---------|----------------| | `FOUND` | Messages found | Process messages, continue loop | | `NO_NEW_MSG` | No new messages at end of queue | `break` - end of queue, correct | | `NO_MATCHED_MSG` | Messages exist but none matched tag filter | `break` - correct for tag-filtered scans | | `OFFSET_ILLEGAL` | Requested offset is stale/invalid (messages cleaned up, broker restarted, offset before min offset) | **`continue`** with `nextBeginOffset` - the broker returns a corrected offset that should be retried | The current code treats `OFFSET_ILLEGAL` the same as `NO_NEW_MSG`, breaking the scan. But `OFFSET_ILLEGAL` does **not** mean "no more messages" - it means "the offset you requested is no longer valid, here is a corrected one." The corrected offset in `nextBeginOffset` points to the earliest still-available message, and the pull should be retried from there. ### When does OFFSET_ILLEGAL occur in practice? - **Message retention cleanup**: RocketMQ deletes expired message files based on `fileReservedTime` (default 72h). If the query time range starts before the earliest available message, `searchOffset(queue, begin)` returns an offset that has already been deleted. The first pull returns `OFFSET_ILLEGAL`. - **Broker restart / abnormal shutdown**: After a crash or restart, the broker may rebuild its ConsumeQueue, and previously valid offsets can become illegal. - **Newly created queues**: A queue with no messages yet may return `OFFSET_ILLEGAL` for offset 0 in certain broker versions. - **DLQ topics**: Dead-letter topics are low-throughput and often have gaps in offsets after retention cleanup. The DLQ resend scan is particularly susceptible. In all these cases, the queue may still contain **valid messages after the corrected offset**, but the current code never reaches them. ### Steps to Reproduce 1. Start a RocketMQ cluster with `fileReservedTime=1` (1 hour retention). 2. Produce 1000 messages to a topic. 3. Wait 2 hours (all messages are now expired and cleaned up). 4. Produce 500 new messages to the same topic. 5. In RocketMQ Studio, query the topic with a time range that starts **before** the cleanup (e.g., 3 hours ago to now). 6. Observe: the query returns **zero results** because `searchOffset` returns a stale offset, the first pull returns `OFFSET_ILLEGAL`, and the scan breaks immediately. 7. Verify: query with a time range that starts **after** the cleanup (e.g., 30 minutes ago to now) - the 500 new messages are found. ### What Did You Expect to See? The scan should recover from `OFFSET_ILLEGAL` by retrying from `nextBeginOffset`, and eventually find all 500 newer messages even when the query range starts before the retention boundary. ### What Did You See Instead? Zero results - the scan breaks on the first `OFFSET_ILLEGAL` pull and never reaches the valid messages. ### Additional Context **Relationship to #1159**: Issue #1159 addresses the case where `nextBeginOffset <= offset` (offset does not advance). This issue is **different**: when `OFFSET_ILLEGAL` is returned, `nextBeginOffset > offset` (the offset DOES advance to a corrected value), but the code breaks before retrying the pull from that corrected offset. **Affected files and methods**: - `server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java` - `queryByTopic()` at approximately line 213 - `server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java` - `collectDeadLetters()` at approximately line 195 Both methods share the exact same scan pattern and have the identical bug. ### Fix suggestion Replace the blanket `break` on non-FOUND status with status-specific handling: ```java offset = nextOffset; if (pullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL) { // The broker returned a corrected offset in nextBeginOffset. // Retry from the corrected position instead of abandoning the queue. log.debug("Offset {} was illegal for queue {} in {}, retrying from {}", offset, queue, topicOrDlq, nextOffset); continue; } if (pullResult.getPullStatus() != PullStatus.FOUND || pullResult.getMsgFoundList() == null) { // NO_NEW_MSG or NO_MATCHED_MSG: end of queue, stop scanning. break; } ``` Additionally, a safety guard should be added to prevent infinite loops if the broker repeatedly returns `OFFSET_ILLEGAL` without making progress (e.g., a max-retry counter per queue, defaulting to 3 consecutive `OFFSET_ILLEGAL` results before breaking). This fix adds approximately 15-20 lines per method (status branching + retry guard) across two files, totaling ~35-40 lines of new code. No existing logic is deleted - the `NO_NEW_MSG` and `NO_MATCHED_MSG` break paths are preserved exactly as before. -- 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]
