lucasbru commented on code in PR #22608:
URL: https://github.com/apache/kafka/pull/22608#discussion_r3437029914
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/MockChangelogReader.java:
##########
@@ -73,6 +73,11 @@ public Set<TopicPartition> completedChangelogs() {
return restoringPartitions;
}
+ @Override
+ public Map<TopicPartition, Long> logicalChangelogEndOffsets() {
Review Comment:
empty? shouldn't it be mockable
##########
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:
So we are setting `changelogMetadata.restoreEndOffset` here which wasn't set
before. Have you checked that there are no side-effect from this?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -442,6 +453,54 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(final
}
}
+ private void updateTaskOffsetSumSnapshot() {
+ final Map<StreamsRebalanceData.TaskId, Long> offsetSnapshot = new
HashMap<>(updatingTasks.size());
+ final Map<StreamsRebalanceData.TaskId, Long> endOffsetSnapshot =
new HashMap<>(updatingTasks.size());
+
+ for (final Task task : updatingTasks.values()) {
+ long sum = 0L;
+ boolean unknownOffsetFound = false;
+ for (final Long offset : task.changelogOffsets().values()) {
+ if (offset == null || offset == OFFSET_UNKNOWN) {
+ unknownOffsetFound = true;
+ // need to continue, as we want MAX_VALUE (overflow)
"win" over 0 (unknown)
+ // if we overflow, it doesn't matter that we didn't
know an offset
+ continue;
+ }
+ if (sum > Long.MAX_VALUE - offset) {
+ sum = Long.MAX_VALUE;
+ break;
+ }
+ sum += offset;
+ }
+ if (unknownOffsetFound && sum != Long.MAX_VALUE) {
+ sum = 0;
+ }
+
+ final Map<TopicPartition, Long> changelogEndOffsets =
changelogReader.logicalChangelogEndOffsets();
Review Comment:
move out of the loop?
--
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]
