Copilot commented on code in PR #3356:
URL: https://github.com/apache/fluss/pull/3356#discussion_r3298163203
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java:
##########
@@ -65,26 +65,30 @@ public class FlinkSourceReaderMetrics {
// Map for tracking current consuming offsets
private final Map<TableBucket, Long> offsets = new HashMap<>();
- // For currentFetchEventTimeLag metric
- private volatile long currentFetchEventTimeLag = UNINITIALIZED;
+ private volatile long maxFetchEventTimeLag = UNINITIALIZED;
+ // Map for tracking current fetch event time lag by table bucket
+ private final Map<TableBucket, Long> currentFetchEventTimeLags = new
HashMap<>();
public FlinkSourceReaderMetrics(SourceReaderMetricGroup
sourceReaderMetricGroup) {
this.sourceReaderMetricGroup = sourceReaderMetricGroup;
this.flussSourceReaderMetricGroup =
sourceReaderMetricGroup.addGroup(FLUSS_METRIC_GROUP).addGroup(READER_METRIC_GROUP);
}
- public void reportRecordEventTime(long lag) {
- if (currentFetchEventTimeLag == UNINITIALIZED) {
- // Lazily register the currentFetchEventTimeLag
- // Set the lag before registering the metric to avoid metric
reporter getting
- // the uninitialized value
- currentFetchEventTimeLag = lag;
- sourceReaderMetricGroup.gauge(
- MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, () ->
currentFetchEventTimeLag);
- return;
+ public void reportRecordEventTime(TableBucket tableBucket, long timestamp)
{
+ if (!currentFetchEventTimeLags.containsKey(tableBucket)) {
+ registerEventTimeLagMetricsForTableBucket(tableBucket);
+ }
+ long lag = System.currentTimeMillis() - timestamp;
+ currentFetchEventTimeLags.put(tableBucket, lag);
+
+ if (lag > maxFetchEventTimeLag) {
+ if (maxFetchEventTimeLag == UNINITIALIZED) {
+ sourceReaderMetricGroup.gauge(
+ MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, () ->
maxFetchEventTimeLag);
+ }
+ maxFetchEventTimeLag = lag;
}
Review Comment:
`maxFetchEventTimeLag` is only updated when the new lag is larger, so
`CURRENT_FETCH_EVENT_TIME_LAG` becomes a monotonic “max-ever” gauge. That
doesn’t match the metric name/previous behavior (it should represent the
current max lag across buckets) and will stay artificially high after the
lagging bucket catches up. Consider recomputing the max from
`currentFetchEventTimeLags` on each update (or maintaining a correct max that
can decrease when the current-max bucket improves).
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetricsTest.java:
##########
@@ -64,6 +65,61 @@ void testCurrentOffsetTracking() {
assertCurrentOffset(t3, 15513L, metricListener);
}
+ @Test
+ void testCurrentFetchEventTimeLagTracksMaxLag() {
+ MetricListener metricListener = new MetricListener();
+ FlinkSourceReaderMetrics flinkSourceReaderMetrics =
+ new FlinkSourceReaderMetrics(
+
InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
+ TableBucket tableBucket0 = new TableBucket(0, 0);
+ TableBucket tableBucket1 = new TableBucket(0, 1);
+
+ long timestamp = System.currentTimeMillis() - 100000L;
+ flinkSourceReaderMetrics.reportRecordEventTime(tableBucket0,
timestamp);
+
+ Optional<Gauge<Long>> readerEventTimeLagGauge =
+
metricListener.getGauge(MetricNames.CURRENT_FETCH_EVENT_TIME_LAG);
+ Optional<Gauge<Long>> bucket0EventTimeLagGauge =
+ metricListener.getGauge(
+ FLUSS_METRIC_GROUP,
+ READER_METRIC_GROUP,
+ BUCKET_GROUP,
+ String.valueOf(tableBucket0.getBucket()),
+ MetricNames.CURRENT_FETCH_EVENT_TIME_LAG);
+ assertThat(readerEventTimeLagGauge).isPresent();
+ assertThat(bucket0EventTimeLagGauge).isPresent();
+ long readerEventTimeLag = readerEventTimeLagGauge.get().getValue();
+ long bucket0EventTimeLag = bucket0EventTimeLagGauge.get().getValue();
+
+ flinkSourceReaderMetrics.reportRecordEventTime(tableBucket0, timestamp
- 100000L);
+ long maxReaderEventTimeLag = readerEventTimeLagGauge.get().getValue();
+ assertThat(maxReaderEventTimeLag).isGreaterThan(readerEventTimeLag);
+ assertThat((long) bucket0EventTimeLagGauge.get().getValue())
+ .isGreaterThan(bucket0EventTimeLag);
+
+ long newerTimestamp = System.currentTimeMillis();
+ flinkSourceReaderMetrics.reportRecordEventTime(tableBucket1,
newerTimestamp);
+ Optional<Gauge<Long>> bucket1EventTimeLagGauge =
+ metricListener.getGauge(
+ FLUSS_METRIC_GROUP,
+ READER_METRIC_GROUP,
+ BUCKET_GROUP,
+ String.valueOf(tableBucket1.getBucket()),
+ MetricNames.CURRENT_FETCH_EVENT_TIME_LAG);
+ assertThat(bucket1EventTimeLagGauge).isPresent();
+ assertThat((long) readerEventTimeLagGauge.get().getValue())
+ .isEqualTo(maxReaderEventTimeLag);
+ assertThat((long) bucket1EventTimeLagGauge.get().getValue())
+ .isLessThan(maxReaderEventTimeLag);
+
+ long updatedBucket0Timestamp = newerTimestamp - 50000L;
+ flinkSourceReaderMetrics.reportRecordEventTime(tableBucket0,
updatedBucket0Timestamp);
+ assertThat((long) readerEventTimeLagGauge.get().getValue())
+ .isEqualTo(maxReaderEventTimeLag);
+ assertThat((long) bucket0EventTimeLagGauge.get().getValue())
+ .isLessThan(maxReaderEventTimeLag);
Review Comment:
The assertions here expect the reader-level `CURRENT_FETCH_EVENT_TIME_LAG`
gauge to remain equal to the previous maximum even after a newer timestamp is
reported for another bucket. If the intended meaning is “current max lag across
buckets”, the gauge should be able to decrease when the lagging bucket catches
up; otherwise the test is locking in a monotonic/max-ever behavior.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java:
##########
@@ -65,26 +65,30 @@ public class FlinkSourceReaderMetrics {
// Map for tracking current consuming offsets
private final Map<TableBucket, Long> offsets = new HashMap<>();
- // For currentFetchEventTimeLag metric
- private volatile long currentFetchEventTimeLag = UNINITIALIZED;
+ private volatile long maxFetchEventTimeLag = UNINITIALIZED;
+ // Map for tracking current fetch event time lag by table bucket
+ private final Map<TableBucket, Long> currentFetchEventTimeLags = new
HashMap<>();
Review Comment:
`currentFetchEventTimeLags` is a plain `HashMap`, but its values are exposed
via Flink gauges (read by metric reporters, typically on a different thread)
while being mutated in `reportRecordEventTime(...)`. This can cause data
races/undefined behavior; consider switching to `ConcurrentHashMap` (or storing
per-bucket `AtomicLong`s) and ensuring `maxFetchEventTimeLag` updates are
thread-safe as well.
--
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]
