KAFKA-20711: Measure restore-remaining-records in offset slots#22614
KAFKA-20711: Measure restore-remaining-records in offset slots#22614nicktelford wants to merge 1 commit into
Conversation
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Thanks for the fix.
On thing I am wondering is about restore-total and restore-rate metric. My understanding is, that both would measure how many actual records we did restore. So if we add TX-markers and offset-gaps (from compaction), it seems these metrics would not become incorrect?
There a point about keeping restore-remaining and restore-total "in sync", but in the end restore-remaining can only be an estimation of how many actually records are remaining. Ideally restore-remaining would tell us how many records are remaining, but this is something we just don't know, and can only estimate via offsets -- but it seem to go a step too far to "mess" with the other two metrics which are correct now.
| final Long previousOffset, | ||
| final long restoredToOffset) { | ||
| final long restoredFrom = previousOffset == null ? changelogMetadata.restoreStartOffset - 1 : previousOffset; | ||
| if (restoredToOffset > restoredFrom) { |
There was a problem hiding this comment.
Why do we need this guard? Should this not be always true? For which case it would be false?
There was a problem hiding this 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?
@mjsax Yeah, I opted to change the semantics of A concrete example here is that if we ensure these metrics all measure the same thing, you can derive an estimated time remaining for restore by calculating It would of course be best for |
|
@nicktelford -- I was thinking about this more, and also discussed with other. -- I don't think we should tightly couple Looking into the code, it seems both are already captured by independent We might also want to update the description of both metrics, to call out the difference, and that one is exact, while the other is an estimation. |
@mjsax the problem with having those metrics record subtly different things is that you can't combine them to determine "time remaining" = restore-remaining / restore-rate. This, for me at least, is the killer feature of these metrics. More generally, users don't have visibility into consumable records vs. transaction markers/aborted records on a Kafka topic: we just see offsets. All other Kafka topic metrics are based on log offsets, not number of consumable records, so this is a bit of an outlier. To provide a hypothetical, pathological example: it's possible for a topic to contain a large number of aborted transactions, compacted-away records, transaction markers etc. and very few real records. If this happens, users would see their application is "restoring", but then see the restore-rate as much lower than they would expect. Another perspective: when consumers use a regular Kafka consumer group, the consumed offsets are stored on the broker, and made available via metrics. This enables users to calculate things like lag, consumption rate, etc. from just the broker metrics, and these metrics are entirely offset-based. However, since the restore consumer doesn't use a consumer group (it stores its offsets inside StateStores) the only metrics available to users are the I understand why you're unwilling to change the semantics of existing metrics: perhaps the answer is to create some new metrics that track offsets instead of restored records? |
|
Well, consumers do already exposes While your example with aborted TX and TX markers is "valid", the same problem exist for log-compaction with the main difference that many offsets don't exist any longer. The log compaction case is the more common one (I would only expect a low percentage of aborted TX), and if we would count non-existing offset as part of restore rate, it could be very off, and it might also vary a lot (ie, older segments with more compaction would have more missing offsets, while newer segment might not be compacted yet, appearing to restore slower, even if they are not, if we look at offsets).
Maybe, but for compacted topic this approach seems to be broken? In the end, |
The restore-remaining-records-total metric is initialized from the changelog offset range (restoreEndOffset - startOffset), which counts every offset slot, but it was decremented by the number of records actually restored. These diverge whenever the changelog contains offsets the restore consumer never returns as records -- transaction markers or compacted-away records -- leaving the metric stuck above zero after restoration completed. The metric is now decremented in offset slots: each batch advances it by the change in store offset, and on completion any trailing slots between the last restored record and the end offset are accounted for, so it lands at exactly zero. Task#recordRestoration takes the offset-slot count separately from the record count so that restore-total and restore-rate continue to measure records restored. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
nicktelford
changed the title
803a85f to
3ad3020
Compare
|
@mjsax I've made the requested changes and rebased against the latest trunk. You mention |
2 participants
The
restore-remaining-records-totaltask metric overshoots and neverreaches zero under EOS (and, more mildly, on compacted changelogs). The
metric is initialized in
prepareChangelogsfrom the offset rangerestoreEndOffset - startOffset, which counts every offset slot —including those occupied by transaction markers and
aborted/compacted-away records. It was then decremented in
restoreChangelogby the number of records actually restored(
bufferedLimitIndex). Because theread_committedrestore consumernever returns markers or aborted records (though its position advances
past them), the sum of the decrements is always smaller than the initial
value by exactly the number of those offset slots, so the metric stays
stuck at a positive value even after restoration completes.
The fix makes the decrements consistent with the initialization by
expressing the remaining-records metric's progress in changelog offset
slots rather than record counts. Each batch advances the metric by the
change in the store's offset (
storeMetadata.offset(), whichrestore()sets to the last restored record's offset), and oncompletion any trailing slots between the last restored record and the
end offset are accounted for, so the metric lands at exactly zero. A new
restoreStartOffsetfield records the consumer position at whichrestoration began, so the first batch is measured from the true start
(covering the from-the-beginning case and any leading marker offsets).
Task#recordRestorationnow takes the offset-slot count separately fromthe record count: the offset-slot count drives
restore-remaining-records-total, whilerestore-total/restore-rate(and the standby
update-total/update-rate) measure the recordsactually restored. The per-batch listener callbacks
(
onBatchRestored/onBatchLoaded) andtotalRestoredlikewise reporttrue record counts.
A unit test reproduces the gap (a changelog whose consumer position
advances past offsets that are never returned as records) and asserts
the metric decrements to exactly zero on completion.
🤖 Generated with Claude Code
Reviewers: Matthias J. Sax matthias@confluent.io