mjsax commented on code in PR #22614:
URL: https://github.com/apache/kafka/pull/22614#discussion_r3692796176
##########
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,
Review Comment:
```suggestion
final Long lastRestoredOffset, //
this is not a "position" so we need to correct it below
```
##########
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:
This was not addressed yet -- while we need to keep `storeMetadata.offset`
as-is, due to `null` handling, I would still think we should pass
`changelogMetadata.restoreEndOffset` as last parameter, not subtracting one.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -680,9 +685,9 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
changelogMetadata.bufferedRecords.clear();
}
- task.recordRestoration(time, numRecords, false);
-
final Long currentOffset = storeMetadata.offset();
+ recordRestorationProgress(task, changelogMetadata, numRecords,
offsetBeforeRestore, currentOffset);
Review Comment:
```suggestion
recordRestorationProgress(task, changelogMetadata, numRecords,
offsetBeforeRestore, currentOffset + 1);
```
We need to pass in a position here -- `currentOffset` cannot be `null` as we
did complete a restore above, so this should be save.
##########
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:
Seems we need to keep this as-is, due to `null` handling...
##########
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:
Seems this was not addressed yet? We should define `restoredFrom` as a
position, and thus it should be:
```
final long restoredFrom = previousOffset == null ?
changelogMetadata.restoreStartOffset : previousOffset + 1;
```
(mind the suggested renaming from `previousOffset` to `lastRestoredOffset`)
In combination with my other comment, to pass in a "position" as last
parameter `restoredToOffset`, it evens out, but is now position based and
easier to understand.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -680,9 +685,9 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
changelogMetadata.bufferedRecords.clear();
}
- task.recordRestoration(time, numRecords, false);
-
final Long currentOffset = storeMetadata.offset();
Review Comment:
```suggestion
final long currentOffset = storeMetadata.offset();
```
This should never return `null` so we should make it explicit by using a
primitive type (we could also add a comment about it)
--
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]
