[ISSUE #10076] Make orderly resetOffset wait on consume lock while preserving timeout semantics#10175
Merged
lizhimins merged 1 commit intoapache:developfrom Mar 31, 2026
Conversation
lizhimins
approved these changes
Mar 31, 2026
| if (topic.equals(mq.getTopic()) && offset != null) { | ||
| try { | ||
| ProcessQueue pq = processQueueTable.get(mq); | ||
| waitResetOffsetReady(consumer, pq); |
Member
There was a problem hiding this comment.
Note: The purpose of acquiring a write lock is to wait for the currently ongoing sequential consumption process to complete.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which Issue(s) This PR Fixes
Brief Description
This change mainly optimizes the execution of resetOffset on the client side for orderly consumption to reduce the waiting time (during this period, the consumer is in a suspended state and should be resumed as soon as possible).
For orderly consumption, resetOffset no longer enforces a mandatory 10-second wait. Instead, it reuses the writeLock of the ProcessQueue. If the lock is successfully acquired, it indicates that no other thread is currently holding the ProcessQueue, and subsequent operations can be safely executed. Otherwise, it waits for the lock to be released (up to a maximum of 10 seconds, which is consistent with the original logic).
For concurrent consumption, the logic currently remains the same as before, i.e., waiting for 10 seconds. The current idea is to define a counter variable in the ProcessQueue to track the number of threads currently holding it. However, this approach would be intrusive to the codebase and require modifications to ConsumeMessageConcurrentlyService.
How Did You Test This Change?