mjsax commented on code in PR #22608:
URL: https://github.com/apache/kafka/pull/22608#discussion_r3439613565
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -713,6 +731,21 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
return numRecords;
}
+ // For non-source-topic standby changelogs, the reader never populates
restoreEndOffset
+ // through initializeChangelogs /
maybeUpdateLimitOffsetsForStandbyChangelogs. Refresh it
+ // here from data we already have on hand (the consumer's cached Fetch
high-water-mark),
+ // so taskEndOffsetSumSnapshot can report the changelog's log-end-offset
for warm-up lag
+ // without issuing a new RPC.
+ private void maybeRefreshNonSourceStandbyEndOffset(final ChangelogMetadata
changelogMetadata,
+ final TopicPartition
partition,
+ final OptionalLong
optionalLag) {
+ if (optionalLag.isPresent()
+ && changelogMetadata.stateManager.taskType() ==
TaskType.STANDBY
+ &&
!changelogMetadata.stateManager.changelogAsSource(partition)) {
Review Comment:
We actually have a failing test on this build, triggered by this change... 😬
--
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]
