apoorvmittal10 commented on code in PR #23892:
URL: https://github.com/apache/kafka/pull/23892#discussion_r4339954393
##########
core/src/main/java/kafka/server/share/DelayedShareFetch.java:
##########
@@ -850,6 +860,61 @@ CompletableFuture<LinkedHashMap<TopicIdPartition,
LogReadResult>> combineLogRead
});
}
+ /**
+ * Returns the record budget left for further reads once the already
fetched data has been handed out,
+ * or 0 if the budget is exhausted. Records are counted by offset span,
from the fetch offset to the last
+ * offset of the last batch, which is the same approximation {@link
SharePartition#acquire} makes when it
+ * bounds the acquired records by max fetch records. The span starts at
the fetch offset rather than at
+ * the base offset of the first batch because the fetched data starts at a
batch boundary, and the records
+ * of that batch prior to the fetch offset are already in-flight or
acknowledged, hence are not handed out
+ * again and must not consume the budget.
+ * <ul>
+ * <li>{@link RecordBatch#lastOffset()} is called at most once per
partition, on the last batch, rather
+ * than on every batch. For file backed records the base offset comes
from the bytes already read while
+ * iterating, whereas the last offset may require loading the batch
header from the file channel.</li>
+ * <li>Iteration stops as soon as the base offsets alone prove the
budget is spent. Base offsets are
+ * strictly increasing, so the offsets between the start of the span
and a batch's base offset are
+ * already accounted for and the batch itself holds at least one more
record.</li>
+ * </ul>
+ *
+ * @param maxFetchRecords The record budget of the share fetch request.
+ * @param fetchOffsets The fetch offsets of the acquired topic partitions.
+ * @param fetchedData The already fetched data, keyed by topic partition.
+ * @return The remaining record budget, or 0 if none remains.
+ */
+ private static int remainingRecordBudget(int maxFetchRecords,
+ Map<TopicIdPartition, Long>
fetchOffsets,
+ Map<TopicIdPartition,
LogReadResult> fetchedData) {
+ long remaining = maxFetchRecords;
+ for (Map.Entry<TopicIdPartition, LogReadResult> entry :
fetchedData.entrySet()) {
+ FetchDataInfo fetchDataInfo = entry.getValue().info();
+ if (fetchDataInfo == null || fetchDataInfo.records == null) {
+ continue;
+ }
+
+ Records records = fetchDataInfo.records;
+ // The span starts at the fetch offset, or at the base offset of
the first batch if that is
+ // higher or the fetch offset is unknown.
+ long startOffset = fetchOffsets.getOrDefault(entry.getKey(), -1L);
Review Comment:
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.
--
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]
