apoorvmittal10 commented on code in PR #18804:
URL: https://github.com/apache/kafka/pull/18804#discussion_r1946814224
##########
core/src/main/java/kafka/server/share/ShareFetchUtils.java:
##########
@@ -187,4 +185,78 @@ static Partition partition(ReplicaManager replicaManager,
TopicPartition tp) {
}
return partition;
}
+
+ /**
+ * Slice the fetch records based on the acquired records. The slicing is
done based on the first
+ * and last offset of the acquired records from the list. The slicing
doesn't consider individual
+ * acquired batches rather the boundaries of the acquired list.
+ *
+ * @param records The records to be sliced.
+ * @param shareAcquiredRecords The share acquired records containing the
non-empty acquired records.
+ * @return The sliced records, if the records are of type FileRecords and
the acquired records are a subset
+ * of the fetched records. Otherwise, the original records are
returned.
+ */
+ static Records maybeSliceFetchRecords(Records records,
ShareAcquiredRecords shareAcquiredRecords) {
+ if (!shareAcquiredRecords.subsetAcquired() || !(records instanceof
FileRecords fileRecords)) {
+ return records;
+ }
+ // The acquired records should be non-empty, do not check as the
method is called only when the
+ // acquired records are non-empty.
+ List<AcquiredRecords> acquiredRecords =
shareAcquiredRecords.acquiredRecords();
+ try {
+ final long firstAcquiredOffset =
acquiredRecords.get(0).firstOffset();
+ final long lastAcquiredOffset =
acquiredRecords.get(acquiredRecords.size() - 1).lastOffset();
+ int startPosition = 0;
+ int size = 0;
+ // Track the previous batch to adjust the start position in case
the first acquired offset
+ // is between the batch.
+ FileChannelRecordBatch previousBatch = null;
+ for (FileChannelRecordBatch batch : fileRecords.batches()) {
+ // If the batch base offset is less than the first acquired
offset, then the start position
+ // should be updated to skip the batch.
+ if (batch.baseOffset() < firstAcquiredOffset) {
Review Comment:
Yes you are correct that would have been the easiest way. But as lastOffset
of batch loads header hence I have avoided that call by maintaining
`previousBatch`.
--
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]
