nicktelford commented on code in PR #22614:
URL: https://github.com/apache/kafka/pull/22614#discussion_r3473301161
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -713,6 +730,28 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
return numRecords;
}
+ /**
+ * Advance the restore/update total and rate metrics (and, for active
tasks, the remaining-records
+ * metric) by the number of changelog offset slots covered between {@code
previousOffset} and
+ * {@code restoredToOffset}, rather than by a record count. Measuring in
offset slots accounts for
+ * offsets occupied by transaction markers or compacted-away records that
the restore consumer never
+ * returns, keeping the metrics consistent with the offset-based
remaining-records value so it reaches
+ * exactly zero on completion.
+ *
+ * @param previousOffset the store offset before this progress (null if
nothing has been restored yet,
+ * in which case restoration is measured from
{@code restoreStartOffset})
+ * @param restoredToOffset the offset restoration has now reached
+ */
+ private void recordRestorationProgress(final Task task,
+ final ChangelogMetadata
changelogMetadata,
+ final Long previousOffset,
+ final long restoredToOffset) {
+ final long restoredFrom = previousOffset == null ?
changelogMetadata.restoreStartOffset - 1 : previousOffset;
+ if (restoredToOffset > restoredFrom) {
Review Comment:
This is to defend against `previousOffset == restoredToOffset`. Maybe this
is overly defensive. I think this _might_ be possible if the `StateStore` has a
committed offset > `restoredToOffset`, but I don't think that's possible, since
the committed offset will dictate where the restore begins.
Would you like me to remove this?
--
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]
