[apoorvmittal10]

On the slow path, DelayedShareFetch.combineLogReadResponse reuses the
data read in tryComplete and issues a second readFromLog for
partitions that became acquirable only at completion. The second read
was issued even when the already fetched data covered maxFetchRecords,
in which case SharePartition.acquire receives a zero budget and
returns 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

[adixitconfluent]

Thanks for the PR. Mostly looks good to me. One question

[adixitconfluent]

what is the case when startOffset is not found in fetchOffsets?

[apoorvmittal10]

This is just defensive code, ideally fetchOffset should not be empty and in case it is then it anyways gets reset while comparing against baseOffset.

[adixitconfluent]

okay, I was thinking maybe if such scenario arrives, we should put a WARNING log and return, but if you think that this is fine, we can go ahead with this way as well

[AndrewJSchofield]

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.

[apoorvmittal10]

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.

Yes it's correct that unfortunately that used to happen. But it was not intended, correct me if I am wrong @adixitconfluent. We make best effort basis to acquire and process the partitions in order of partitions in the request. This additional case shall trigger when some partitions are fetched but yet some additional are needed later. However all topics are already acquired. The order is anyways not maintained in this latter case so this change should ideally not cause regression.

Prior calling the delayedsharefetch, partitions are always rotated in share partition manager to prevent starvation.

[apoorvmittal10]

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.

Yes it's correct that unfortunately that used to happen. But it was not intended, correct me if I am wrong @adixitconfluent. We make best effort basis to acquire and process the partitions in order of partitions in the request. This additional case shall trigger when some partitions are fetched but yet some additional are needed later. However all topics are already acquired. The order is anyways not maintained in this latter case so this change should ideally not cause regression.

Prior calling the delayedsharefetch, partitions are always rotated in share partition manager to prevent starvation.

I (claude) wrote a test and can see starvation of 1 partition. It was just a matter of sequence of operations that this piece was earlier working, @adixitconfluent correct me if I am wrong, as I do not think we deliberately put the missing partitions prior to already fetched in the response. The problem is a bit deep while updating fetchOffsetMetadata for the partitions. When all fetched records are not sent to client then fetchOffsetMetadata is also bumped for trimmed partitions which extrensically adds the partitions to the missing list. @AndrewJSchofield @adixitconfluent Let me know the choice, I can address that in the same PR or separate PR or prior to current PR (my preference).

[adixitconfluent]

I (claude) wrote a test and can see starvation of 1 partition. It was just a matter of sequence of operations that this piece was earlier working, @adixitconfluent correct me if I am wrong, as I do not think we deliberately put the missing partitions prior to already fetched in the response. The problem is a bit deep while updating fetchOffsetMetadata for the partitions. When all fetched records are not sent to client then fetchOffsetMetadata is also bumped for trimmed partitions which extrensically adds the partitions to the missing list. @AndrewJSchofield @adixitconfluent Let me know the choice, I can address that in the same PR or separate PR or prior to current PR (my preference).

Yes, while writing the code for combineLogReadResponse, I did not really consider starvation but the sequence of operations ensured that there is no starvation and hence we never faced a problem. I think we should fix the problem when all the fetched records are not sent to client then we shouldn't update fetchOffsetMetadata for the unsent partitions. I am okay if that is in a separate PR or if you want me to take it up.

[AndrewJSchofield]

@apoorvmittal10 I think the starvation needs to be solved, either in this PR or a PR which goes before this one.