apoorvmittal10 commented on code in PR #19010:
URL: https://github.com/apache/kafka/pull/19010#discussion_r1969396237
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -1061,33 +1100,105 @@ 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 floorOffset = fetchOffset;
+ Map.Entry<Long, InFlightBatch> floorEntry =
cachedState.floorEntry(fetchOffset);
+ if (floorEntry != null && floorEntry.getValue().lastOffset() >=
fetchOffset) {
Review Comment:
To validate if really the `fetchOffset` is within the floorEntry batch. Say
if cache holds [0-5], [10-11] and fetchOffset is 8, then the floorEntry will be
of [0-5]. In that case, we should not update floorOffset as the floorEntry
batch is prior fetchOffset.
I understand that there might be a question regarding whether such scenario
can happen? Well, not that I could think of i.e. the fetchOffset should be
either at the end or anywhere within the cache where an offset is marked
available. We shall ideally get a specific batch which should contain the
fetchOffset. However, this seems to be a right and safe check for now where the
processing would be deterministic and should not fetch unneccessary batches.
--
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]
