mjsax commented on code in PR #22614:
URL: https://github.com/apache/kafka/pull/22614#discussion_r3606234170
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -730,6 +739,23 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
return numRecords;
}
+ /**
+ * Record restoration progress: restore-total/restore-rate advance by the
records restored
+ * ({@code numRecords}), while the remaining-records metric is decremented
by the offset slots
+ * between {@code previousOffset} (or {@code restoreStartOffset} if null)
and {@code restoredToOffset}.
+ * Measuring the latter in offset slots accounts for offsets the restore
consumer never returns
+ * (transaction markers, compacted records) so it reaches exactly zero on
completion.
+ */
+ private void recordRestorationProgress(final Task task,
+ final ChangelogMetadata
changelogMetadata,
+ final long numRecords,
+ final Long previousOffset,
+ final long restoredToOffset) {
+ final long restoredFrom = previousOffset == null ?
changelogMetadata.restoreStartOffset - 1 : previousOffset;
Review Comment:
Why do we need `-1` and not just `changelogMetadata.restoreStartOffset` --
`restoreStartOffset` should be the "start position" ie, offset of the first
record we restored, what look correct to me.
--
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]
