mjsax commented on code in PR #22608:
URL: https://github.com/apache/kafka/pull/22608#discussion_r3439597592
##########
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:
Not sure what you exactly mean? Can you elaborate? I did not check for any
side-effect, can no idea what you have in mind, and what I could check for?
--
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]
