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.)
```java
// 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/ProcessorStateManager.java:
##########
@@ -533,6 +535,31 @@ void restore(final StateStoreMetadata storeMetadata, final
List<ConsumerRecord<b
}
}
+ /**
+ * Records that restoration has conclusively reached {@code
nextOffsetToFetch} (Kafka next-offset-to-fetch
+ * semantics: log-end offset / restore boundary), converting to
last-applied as
+ * {@code nextOffsetToFetch - 1}. Used when a catch-up boundary is
confirmed with no further records
+ * to apply (KAFKA-14302). Does not change {@link #changelogOffsets()}
mapping of {@code null -> 0}.
+ */
Review Comment:
`nextOffsetToFetch` isn't always the log-end offset — for a changelog
piggy-backed on a source topic it's `min(log-end-offset, committed-offset)`
(see the `restoreEndOffset` field doc), i.e. the committed limit. Maybe drop
the "log-end offset" gloss and reword as?
```java
/**
* Advances the store's restored offset once restoration has reached the end
offset (the limit
* offset for standbys): {@code nextOffsetToFetch} is a next offset to fetch
(exclusive end),
* stored as last-applied {@code nextOffsetToFetch - 1} (KAFKA-14302). Never
moves backwards;
* a non-positive value is a no-op, so the stored offset stays non-negative.
*/
```
##########
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.)
```java
// 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.)
```java
/**
* 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.
*/
```
##########
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.
+ */
+ private void maybeAdvanceStandbyRestoredOffset(final ProcessorStateManager
stateManager,
+ final ChangelogMetadata
changelogMetadata,
+ final StateStoreMetadata
storeMetadata,
+ final TopicPartition
partition) {
+ if (!changelogMetadata.bufferedRecords().isEmpty()) {
+ return;
+ }
+ try {
+ final Long restoreEndOffset = changelogMetadata.restoreEndOffset;
+ if (restoreEndOffset == null) {
+ // Dedicated changelog: no restoreEndOffset. lag == 0 with an
empty buffer means
+ // the restore consumer is at LEO and every fetched record has
already been applied.
+ // position() is the next-fetch boundary, not a last-applied
offset.
Review Comment:
Maybe use `log-end offset` here to match the rest of the file (which doesn't
use `LEO`)?
```java
// dedicated changelog (no restoreEndOffset): lag == 0 with an empty buffer
means the
// consumer is at the log-end offset with every fetched record applied;
position() is
// the next-fetch, not last-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.
+ */
+ private void maybeAdvanceStandbyRestoredOffset(final ProcessorStateManager
stateManager,
+ final ChangelogMetadata
changelogMetadata,
+ final StateStoreMetadata
storeMetadata,
+ final TopicPartition
partition) {
+ if (!changelogMetadata.bufferedRecords().isEmpty()) {
+ return;
+ }
+ try {
+ final Long restoreEndOffset = changelogMetadata.restoreEndOffset;
+ if (restoreEndOffset == null) {
+ // Dedicated changelog: no restoreEndOffset. lag == 0 with an
empty buffer means
+ // the restore consumer is at LEO and every fetched record has
already been applied.
+ // position() is the next-fetch boundary, not a last-applied
offset.
+ final OptionalLong lag = restoreConsumer.currentLag(partition);
+ if (lag.isPresent() && lag.getAsLong() == 0L) {
+ stateManager.advanceRestoredOffsetTo(storeMetadata,
restoreConsumer.position(partition));
+ }
+ } else if (restoreEndOffset > 0L) {
+ // Source changelog: restoreEndOffset is the committed-offset
limit. Advertise
+ // only that boundary, never live position past
unapplied/uncommitted records.
Review Comment:
Maybe drop "boundary" (not used elsewhere) in favour of the `limit` term
already in the file?
```java
// source changelog: restoreEndOffset is the committed-offset limit; record
only up to
// that limit, never the live position past unapplied/uncommitted records
```
--
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]
