[flink] Support pendingRecords metric for Fluss source. - #4250
[flink] Support pendingRecords metric for Fluss source.#4250loserwang1024 wants to merge 1 commit into
Conversation
Track log record lag using scanner high watermarks and current fetch offsets, and aggregate the lag across subscribed buckets.
naivedogger
left a comment
naivedogger
left a comment
There was a problem hiding this comment.
Thanks @loserwang1024 for this pr, left some comments, PTAL.
| } | ||
|
|
||
| @Test | ||
| void testRecordsLagAggregation() { |
There was a problem hiding this comment.
Could we extend testPendingRecordsMetric to cover multiple buckets and remove this separate test? That would verify lag aggregation through actual fetching and metric registration rather than manually updating scanner state. We should also preserve the coverage for excluding unassigned buckets from the total lag.
| * that re-created split readers won't register the metric again. | ||
| */ | ||
| public void registerPendingRecordsGauge(Gauge<Long> pendingRecordsGauge) { | ||
| if (!pendingRecordsGaugeRegistered) { |
There was a problem hiding this comment.
With periodic partition discovery enabled, removing all of a subtask's splits after partition deletion can cause Flink to close its idle fetcher without closing the outer source reader. A later partition assignment creates a new split reader that shares the existing FlinkSourceReaderMetrics. This guard then ignores the new scanner's gauge, leaving pendingRecords bound to the old scanner. Could we register the metric once but update its delegate when a new split reader is created, and add a regression test for this lifecycle?
| * that re-created split readers won't register the metric again. | ||
| */ | ||
| public void registerPendingRecordsGauge(Gauge<Long> pendingRecordsGauge) { | ||
| if (!pendingRecordsGaugeRegistered) { |
There was a problem hiding this comment.
Could we update the backing gauge when a split reader is recreated? The current guard keeps pendingRecords bound to the old scanner, so it may remain at 0 even when the new scanner has a backlog.
4 participants
Track log record lag using scanner high watermarks and current fetch offsets, and aggregate the lag across subscribed buckets.
Purpose
Linked issue: close #138
Brief change log
Tests
API and Format
Documentation