lucasbru commented on code in PR #23991:
URL: https://github.com/apache/kafka/pull/23991#discussion_r4421905730
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -800,7 +801,11 @@ private void maybeAdvanceStandbyRestoredOffset(final
ProcessorStateManager state
// the next offset to fetch, not last-applied
final OptionalLong lag = restoreConsumer.currentLag(partition);
if (lag.isPresent() && lag.getAsLong() == 0L) {
- stateManager.advanceRestoredOffsetTo(storeMetadata,
restoreConsumer.position(partition, Duration.ZERO));
+ final long position = restoreConsumer.position(partition,
Duration.ZERO);
+ stateManager.advanceRestoredOffsetTo(storeMetadata,
position);
+ // the end offset is otherwise only learned from a
restored batch, which never comes when
+ // there is nothing to restore; like there, it is the last
offset, -1 for an empty changelog
+ storeMetadata.setEndOffset(position - 1L);
Review Comment:
This also runs for standby changelogs that are paused while the member
restores an active task (`restore()` goes over all RESTORING changelogs, paused
or not). For a paused partition the consumer keeps the last known high
watermark (or last stable offset under read_committed), so `currentLag` can
stay 0 while the active owner keeps writing to the changelog. A warm-up task
that caught up on an empty changelog before the pause would then keep reporting
lag 0 and could be promoted although it is behind. Before this change it
reported `Long.MAX_VALUE` in that case. The batch path has the same staleness
already, so this isn't new in kind, but should we skip this branch when
`restoreConsumer.paused().contains(partition)`?
--
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]
