mjsax commented on code in PR #22614:
URL: https://github.com/apache/kafka/pull/22614#discussion_r3606240374
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -710,6 +715,10 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
log.info("Finished restoring changelog {} to store {} with a total
number of {} records",
partition, storeName, changelogMetadata.totalRestored);
+ // account for any offset slots past the last restored record
(e.g. trailing transaction
+ // markers) so the remaining-records metric reaches exactly zero
on completion
+ recordRestorationProgress(task, changelogMetadata, 0,
storeMetadata.offset(), changelogMetadata.restoreEndOffset - 1);
Review Comment:
Why `-1`
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -668,6 +671,8 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
if (numRecords != 0) {
final List<ConsumerRecord<byte[], byte[]>> records =
changelogMetadata.bufferedRecords.subList(0, numRecords);
+ // where restoration had reached before this batch; null until the
first batch is restored
+ final Long offsetBeforeRestore = storeMetadata.offset();
Review Comment:
Do we need to add `+1` here (depending what we really want to track, or how
to do the math).
`storeMetadata.offset()` return the offset of the last restored record, to
the "restore position" is one larger.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -710,6 +715,10 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
log.info("Finished restoring changelog {} to store {} with a total
number of {} records",
partition, storeName, changelogMetadata.totalRestored);
+ // account for any offset slots past the last restored record
(e.g. trailing transaction
+ // markers) so the remaining-records metric reaches exactly zero
on completion
+ recordRestorationProgress(task, changelogMetadata, 0,
storeMetadata.offset(), changelogMetadata.restoreEndOffset - 1);
Review Comment:
Base on my other comment, wondering if we would need to to
`storeMetadata.offset() + 1` here?
Or change `recordRestorationProgress` instead and add `+1` inside the method
##########
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:
Should this be `previousOffset + 1` ? (cf my other comment)?
##########
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;
+ final long numOffsets = Math.max(restoredToOffset - restoredFrom, 0L);
Review Comment:
Could the `Math.max(...)` guard hide a bug? In the end, if there is no bug,
it should never become negative?
--
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]
