apoorvmittal10 commented on code in PR #19010:
URL: https://github.com/apache/kafka/pull/19010#discussion_r1967817116
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1061,33 +1098,103 @@ void updateCacheAndOffsets(long logStartOffset) {
}
}
+ /**
+ * The method archives the available records in the cached state that are
between the fetch offset
+ * and the base offset of the first fetched batch. This method is required
to handle the compacted
+ * topics where the already fetched batch which is marked re-available,
might not result in subsequent
+ * fetch response from log. Hence, the batches need to be archived to
allow the SPSO and next fetch
+ * offset to progress.
+ *
+ * @param fetchOffset The fetch offset.
+ * @param baseOffset The base offset of the first fetched batch.
+ */
+ private void maybeArchiveStaleBatches(long fetchOffset, long baseOffset) {
+ lock.writeLock().lock();
+ try {
+ // If the fetch happens from within a batch then fetchOffset can
be ahead of base offset else
+ // should be same as baseOffset of the first fetched batch.
Otherwise, we might need to archive
+ // some stale batches.
+ if (cachedState.isEmpty() || fetchOffset >= baseOffset) {
+ // No stale batches to archive.
+ return;
+ }
+
+ // The fetch offset can exist in the middle of the batch. Hence,
find the floor offset
+ // for the fetch offset and then find the sub-map from the floor
offset to the base offset.
+ long mapFetchOffset = fetchOffset;
+ Map.Entry<Long, InFlightBatch> floorOffset =
cachedState.floorEntry(fetchOffset);
Review Comment:
Done.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -651,13 +682,19 @@ public ShareAcquiredRecords acquire(
lock.writeLock().lock();
try {
long baseOffset = firstBatch.baseOffset();
+ // There might be cached batches which are stale due to topic
compaction hence archive them.
Review Comment:
Done.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
