Repository navigation
MINOR: Skip further reads from log for share fetches if max records reached - #23892
apoorvmittal10 wants to merge 2 commits into
Conversation
github-actions
Bot
added
triage
adixitconfluent
left a comment
adixitconfluent
left a comment
There was a problem hiding this comment.
Thanks for the PR. Mostly looks good to me. One question
There was a problem hiding this comment.
what is the case when startOffset is not found in fetchOffsets?
AndrewJSchofield
left a comment
AndrewJSchofield
left a comment
There was a problem hiding this comment.
Thanks for the PR. An AI-assisted review suggests this PR is a regression in behaviour.
The skip drops the partitions that currently have first claim on the budget (DelayedShareFetch.java:849). The new comment says "the already fetched records are handed out first". They are not. The merge at the end of combineLogReadResponse does missingTopicPartitionsLogReadResponse.putAll(existingFetchedData), so the missing partitions sit first in the combined map, and processFetchResponse acquires in map order with maxFetchRecords - acquiredRecordsCount. So before this change the missing partitions were acquired first and the already-fetched ones got the remainder. After it, the already-fetched span is used to decide the missing ones get nothing. Concrete scenario, one consumer, tp0 and tp1 both with deep backlogs:
- tp0 is served, so its next fetch offset moves and its cached fetch-offset metadata goes stale. It is read in tryComplete and lands in the already-fetched set every time.
- tp1 is not served, so its offset and cached metadata stay valid. It is never read in tryComplete and is always a "missing" partition.
- Before the change, tp1 was acquired first on each fetch and the two alternated naturally. After it, tp0's span is at least maxFetchRecords, so tp1 is skipped on every fetch for as long as tp0 has a backlog. That is indefinite starvation of tp1.
There was a problem hiding this comment.
The code actually does opposite of the comment. It puts missing first and then the existing ones
There was a problem hiding this comment.
This is what the AI highlighted too.
3 participants
On the slow path,
DelayedShareFetch.combineLogReadResponsereuses thedata read in
tryCompleteand issues a secondreadFromLogforpartitions that became acquirable only at completion. The second read
was issued even when the already fetched data covered
maxFetchRecords,in which case
SharePartition.acquirereceives a zero budget andreturns empty, so the read was wasted.
The remaining record budget is now computed from the already fetched
data and the second read is skipped when nothing remains. Skipped
partitions are absent from the response, as share sessions already do
for partitions without data, and their locks are released on the same
path as before.
Reviewers: Abhinav Dixit adixit@confluent.io, Andrew Schofield
aschofield@confluent.io