apoorvmittal10 commented on code in PR #17322:
URL: https://github.com/apache/kafka/pull/17322#discussion_r1805429253
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -84,24 +88,31 @@ static Map<TopicIdPartition,
ShareFetchResponseData.PartitionData> processFetchR
partitionData.setErrorMessage(Errors.NONE.message());
}
} else {
- List<AcquiredRecords> acquiredRecords =
sharePartition.acquire(shareFetchData.memberId(), fetchPartitionData);
+ ShareAcquiredRecords shareAcquiredRecords =
sharePartition.acquire(shareFetchData.memberId(),
shareFetchData.maxFetchRecords() - acquiredRecordsCount, fetchPartitionData);
log.trace("Acquired records for topicIdPartition: {} with
share fetch data: {}, records: {}",
- topicIdPartition, shareFetchData, acquiredRecords);
+ topicIdPartition, shareFetchData, shareAcquiredRecords);
// Maybe, in the future, check if no records are acquired, and
we want to retry
// replica manager fetch. Depends on the share partition
manager implementation,
// if we want parallel requests for the same share partition
or not.
- if (acquiredRecords.isEmpty()) {
+ if (shareAcquiredRecords.records().isEmpty()) {
partitionData
.setRecords(null)
.setAcquiredRecords(Collections.emptyList());
} else {
partitionData
+ // We set the records to the fetchPartitionData
records. We do not alter the records
+ // fetched from the replica manager as they follow
zero copy buffer. The acquired records
Review Comment:
Yeah, that's right and hence I wrote this comment. We ll be experimenting
with max bytes and max fetch records to see how less wasteful the response can
be with configurations.
--
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]
