alanlau28 commented on code in PR #23298:
URL: https://github.com/apache/kafka/pull/23298#discussion_r3972814271
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java:
##########
@@ -92,13 +92,15 @@ public static class StateStoreMetadata {
// applied to the store used for both restoration (active and standby
tasks restored offset) and
// normal processing that update stores (written offset); could be
null (when initialized)
//
- // the offset is updated in three ways:
+ // the offset is updated in four ways:
// 1. when loading from the checkpoint file, when the corresponding
task has acquired the state
// directory lock and have registered all the state store; it is
only one-time
// 2. when updating with restore records (by both restoring active
and standby),
// update to the last restore record's offset
// 3. when checkpointing with the given written offsets from record
collector,
// update blindly with the given offset
+ // 4. when restoration reaches a confirmed catch-up boundary with no
further records to apply
+ // (KAFKA-14302), as last-applied = next-fetch - 1, never moving
backwards
Review Comment:
Maybe this can be worded to match the "when X, update to Y" shape of items
1–3, and to note that standbys reach the *limit* offset, not the end offset?
(also dropping the repeated `KAFKA-14302` here — kept once on the
`advanceRestoredOffsetTo` Javadoc.)
```suggestion
// 4. when restoration reaches the end offset (the limit offset
for standbys) with no further
// records to apply, update to next-fetch minus one
(last-applied); never moves backwards
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -749,6 +749,10 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
// markers) so the remaining-records metric reaches exactly zero
on completion
recordRestorationProgress(task, changelogMetadata, 0,
storeMetadata.offset(), changelogMetadata.restoreEndOffset);
+ // Catch-up boundary: advertise restoreEndOffset, not the live
consumer position, which may
+ // already be past records that were buffered and never applied
(KAFKA-14302).
Review Comment:
Maybe this can be worded with the file's own `end offset` term ("catch-up
boundary" isn't used elsewhere) and `record` over `advertise`?
(also dropping the repeated `KAFKA-14302` here — kept once on the
`advanceRestoredOffsetTo` Javadoc.)
```suggestion
// restoration reached the end offset: record restoreEndOffset,
not the consumer position,
// which may already be past buffered records that were never
applied
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -769,6 +775,41 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
return numRecords;
}
+ /**
+ * Advance a standby store's restored offset only at a confirmed catch-up
boundary (KAFKA-14302).
+ * A zero-record poll is not sufficient: retention seeks, source-changelog
buffering, and incomplete
+ * fetches can all leave the consumer at a non-zero position while records
remain to apply.
+ */
Review Comment:
Maybe this can be worded in terms of the end/limit offset rather than
"catch-up boundary" (not a term used elsewhere in the file)?
(also dropping the repeated `KAFKA-14302` here — kept once on the
`advanceRestoredOffsetTo` Javadoc.)
```suggestion
/**
* Advance a standby store's restored offset only once restoration has
actually reached the end
* offset — the limit offset for changelogs piggy-backed on a source
topic. A zero-record poll is
* not sufficient: retention seeks, source buffering, and incomplete
fetches can leave the consumer
* at a non-zero position while records remain to apply.
*/
```
--
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]
