ymwneu opened a new pull request, #11216: URL: https://github.com/apache/rocketmq/pull/11216
…E_FROM_LAST_OFFSET When a consumer using CONSUME_FROM_LAST_OFFSET queries the offset of a brand-new queue that has no stored consumer offset, the broker currently returns QUERY_NOT_FOUND for an empty queue (minOffset <= 0 && checkInMemByConsumeOffset returns false because offset 0 is not yet in memory). The client then falls back to a separate GET_MAX_OFFSET request. If a message is appended to the queue between these two non-atomic requests, the start offset is pinned at 1 and the message at offset 0 is skipped forever. The same checkInMemByConsumeOffset-only logic (and thus the same race) also existed in PopMessageProcessor#getInitOffset and BroadcastOffsetManager#getOffset. Fix: add a shared MessageStore#shouldInitConsumeOffsetToZero(topic, queueId) default method that centralizes the decision: when minOffset is 0, treat an empty queue (maxOffset == 0) as "last offset == 0" and start from 0, or start from 0 when the first message (offset 0) is still in memory; otherwise start from the tail so aged history is not replayed. All three call sites now delegate to it: - ConsumerManageProcessor returns SUCCESS offset=0 instead of QUERY_NOT_FOUND for an empty queue, and swallows ConsumeQueueException to fall back to QUERY_NOT_FOUND (pre-fix behavior) when the queue state cannot be determined. - PopMessageProcessor#getInitOffset and BroadcastOffsetManager#getOffset delegate to the same method (they already propagate ConsumeQueueException). Tests: ConsumerManageProcessorTest drives the shared method's real body across the empty-queue, first-message-in-memory, first-message-on-disk, and max-offset-failure paths; PopMessageProcessorTest and BroadcastOffsetManagerTest each add a case asserting the init offset is pinned to 0 when the first message is still in memory. <!-- Please make sure the target branch is right. In most case, the target branch should be `develop`. --> ### Which Issue(s) This PR Fixes <!-- Please ensure that the related issue has already been created, and [link this pull request to that issue using keywords](<https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword>) to ensure automatic closure. --> - Fixes #11215 ### Brief Description <!-- Write a brief description for your pull request to help the maintainer understand the reasons behind your changes. --> ### How Did You Test This Change? <!-- In order to ensure the code quality of Apache RocketMQ, we expect every pull request to have undergone thorough testing. --> -- 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]
