apoorvmittal10 commented on code in PR #18696:
URL: https://github.com/apache/kafka/pull/18696#discussion_r1935426653
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -641,13 +672,36 @@ public ShareAcquiredRecords acquire(
// The fetched records are already part of the in-flight records.
The records might
// be available for re-delivery hence try acquiring same. The
request batches could
// be an exact match, subset or span over multiple already fetched
batches.
+ long nextBatchStartOffset = baseOffset;
for (Map.Entry<Long, InFlightBatch> entry : subMap.entrySet()) {
// If the acquired count is equal to the max fetch records
then break the loop.
if (acquiredCount >= maxFetchRecords) {
break;
}
InFlightBatch inFlightBatch = entry.getValue();
+
+ // If the initialReadGapOffset window is active, we need to
treat the gaps in between the window as
+ // acquirable. Once the window is inactive (when we have
acquired all the gaps inside the window),
+ // the remaining gaps are natural (data does not exist at
those offsets) and we need nto acquire them.
+ if (isInitialReadGapOffsetWindowActive()) {
+ // If nextBatchStartOffset is less than the key of the
entry, this means the fetch happened for a gap in the cachedState.
+ // Thus, a new batch needs to be acquired for the gap.
+ if (nextBatchStartOffset < entry.getKey()) {
+ ShareAcquiredRecords shareAcquiredRecords =
acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
+ nextBatchStartOffset, entry.getKey() - 1,
batchSize, maxFetchRecords);
+ result.addAll(shareAcquiredRecords.acquiredRecords());
+ acquiredCount += shareAcquiredRecords.count();
+ }
+ // Set nextBatchStartOffset as the last offset of the
current in-flight batch + 1
+ nextBatchStartOffset = inFlightBatch.lastOffset() + 1;
+
Review Comment:
nit: remove line break
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -641,13 +672,36 @@ public ShareAcquiredRecords acquire(
// The fetched records are already part of the in-flight records.
The records might
// be available for re-delivery hence try acquiring same. The
request batches could
// be an exact match, subset or span over multiple already fetched
batches.
+ long nextBatchStartOffset = baseOffset;
Review Comment:
```suggestion
long maybeGapStartOffset = baseOffset;
```
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -641,13 +672,36 @@ public ShareAcquiredRecords acquire(
// The fetched records are already part of the in-flight records.
The records might
// be available for re-delivery hence try acquiring same. The
request batches could
// be an exact match, subset or span over multiple already fetched
batches.
+ long nextBatchStartOffset = baseOffset;
for (Map.Entry<Long, InFlightBatch> entry : subMap.entrySet()) {
// If the acquired count is equal to the max fetch records
then break the loop.
if (acquiredCount >= maxFetchRecords) {
break;
}
InFlightBatch inFlightBatch = entry.getValue();
+
+ // If the initialReadGapOffset window is active, we need to
treat the gaps in between the window as
+ // acquirable. Once the window is inactive (when we have
acquired all the gaps inside the window),
+ // the remaining gaps are natural (data does not exist at
those offsets) and we need nto acquire them.
+ if (isInitialReadGapOffsetWindowActive()) {
+ // If nextBatchStartOffset is less than the key of the
entry, this means the fetch happened for a gap in the cachedState.
+ // Thus, a new batch needs to be acquired for the gap.
+ if (nextBatchStartOffset < entry.getKey()) {
+ ShareAcquiredRecords shareAcquiredRecords =
acquireNewBatchRecords(memberId, fetchPartitionData.records.batches(),
+ nextBatchStartOffset, entry.getKey() - 1,
batchSize, maxFetchRecords);
+ result.addAll(shareAcquiredRecords.acquiredRecords());
+ acquiredCount += shareAcquiredRecords.count();
+ }
+ // Set nextBatchStartOffset as the last offset of the
current in-flight batch + 1
+ nextBatchStartOffset = inFlightBatch.lastOffset() + 1;
+
+ // If the acquired count is equal to the max fetch records
then break the loop.
+ if (acquiredCount >= maxFetchRecords) {
+ break;
+ }
+ }
Review Comment:
@ShivsundarR @AndrewJSchofield Though we always align the acquired batches
atleast on fetched records batch boundairies but this seems to be an exception
in case when initial read gap occurs. Do we see any problem with the approach
in clients parsing of unaligned acquired records with fetched batch? If yes
then we will change the approach here.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -444,6 +450,11 @@ public CompletableFuture<Void> maybeInitialize() {
stateEpoch = partitionData.stateEpoch();
List<PersisterStateBatch> stateBatches =
partitionData.stateBatches();
+ boolean isGapPresentInStateBatches = false;
+ // The previousBatchLastOffset is used to track the last
offset of the previous batch.
+ // For the first batch that should ideally start from
startOffset if there are no gaps,
+ // we assume the previousBatchLastOffset to be startOffset - 1.
+ long previousBatchLastOffset = startOffset - 1;
Review Comment:
I don't think you need to subtract `-1`. It should work without that as
well. StartOffset is 10 and first batch you received is from 10-15. So why to
subtract and track incorrect gap information?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -538,8 +557,20 @@ public long nextFetchOffset() {
}
long nextFetchOffset = -1;
-
+ long gapStartOffset = initialReadGapOffset != null ?
initialReadGapOffset.gapStartOffset() : -1;
Review Comment:
nit: but better to be consistent.
```suggestion
long gapStartOffset = isInitialReadGapOffsetWindowActive() ?
initialReadGapOffset.gapStartOffset() : -1;
```
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -641,13 +672,36 @@ public ShareAcquiredRecords acquire(
// The fetched records are already part of the in-flight records.
The records might
// be available for re-delivery hence try acquiring same. The
request batches could
// be an exact match, subset or span over multiple already fetched
batches.
+ long nextBatchStartOffset = baseOffset;
Review Comment:
Move the variable above `// The fetched records are already...` comments,
and write the comments for the variable.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -641,13 +672,36 @@ public ShareAcquiredRecords acquire(
// The fetched records are already part of the in-flight records.
The records might
// be available for re-delivery hence try acquiring same. The
request batches could
// be an exact match, subset or span over multiple already fetched
batches.
+ long nextBatchStartOffset = baseOffset;
for (Map.Entry<Long, InFlightBatch> entry : subMap.entrySet()) {
// If the acquired count is equal to the max fetch records
then break the loop.
if (acquiredCount >= maxFetchRecords) {
break;
}
InFlightBatch inFlightBatch = entry.getValue();
+
+ // If the initialReadGapOffset window is active, we need to
treat the gaps in between the window as
+ // acquirable. Once the window is inactive (when we have
acquired all the gaps inside the window),
+ // the remaining gaps are natural (data does not exist at
those offsets) and we need nto acquire them.
Review Comment:
```suggestion
// the remaining gaps are natural (data does not exist at
those offsets) and we need not to acquire them.
```
--
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
