Copilot commented on code in PR #3350:
URL: https://github.com/apache/fluss/pull/3350#discussion_r3315215748
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java:
##########
@@ -99,31 +130,165 @@ public void registerTableBucket(TableBucket tableBucket) {
* @param offset Current consuming offset
*/
public void recordCurrentOffset(TableBucket tb, long offset) {
- checkTableBucketTracked(tb);
- offsets.put(tb, offset);
+ BucketMetrics metrics = bucketMetrics.get(tb);
+ if (metrics != null) {
+ metrics.recordCurrentOffset(offset);
+ }
+ }
+
+ public void recordLogEndOffset(TableBucket tb, long logEndOffset) {
+ BucketMetrics metrics = bucketMetrics.get(tb);
+ if (metrics != null) {
+ metrics.updateLogEndOffsetIfAdvanced(logEndOffset);
+ }
+ }
+
+ public void unregisterTableBucket(TableBucket tb) {
+ BucketMetrics metrics = bucketMetrics.get(tb);
+ if (metrics != null) {
+ metrics.deactivate();
+ }
+ }
+
+ public List<SubscribedBucket> subscribedBucketsForLogEndOffsetRefresh() {
+ List<SubscribedBucket> subscribedBuckets = new ArrayList<>();
+ for (Map.Entry<TableBucket, BucketMetrics> entry :
bucketMetrics.entrySet()) {
+ BucketMetrics metrics = entry.getValue();
+ if (metrics.needsLogEndOffsetRefresh()) {
+
subscribedBuckets.add(metrics.toSubscribedBucket(entry.getKey()));
+ }
+ }
+ return subscribedBuckets;
}
// -------- Helper functions --------
- private void registerOffsetMetricsForTableBucket(TableBucket tableBucket) {
+ private void registerRecordsLagMetrics() {
+ flussSourceReaderMetricGroup.gauge(RECORDS_LAG_MAX_METRIC_GAUGE,
this::recordsLagMax);
+ flussSourceReaderMetricGroup.gauge(RECORDS_LAG_SUM_METRIC_GAUGE,
this::recordsLagSum);
+ flussSourceReaderMetricGroup.gauge(LAGGING_BUCKETS_METRIC_GAUGE,
this::laggingBuckets);
+ }
+
+ private void registerOffsetMetricsForTableBucket(
+ TableBucket tableBucket, BucketMetrics bucketMetrics) {
final MetricGroup metricGroup =
tableBucket.getPartitionId() == null
? this.flussSourceReaderMetricGroup
: this.flussSourceReaderMetricGroup.addGroup(
PARTITION_GROUP,
String.valueOf(tableBucket.getPartitionId()));
final MetricGroup bucketGroup =
metricGroup.addGroup(BUCKET_GROUP,
String.valueOf(tableBucket.getBucket()));
- bucketGroup.gauge(
- CURRENT_OFFSET_METRIC_GAUGE,
- () -> offsets.getOrDefault(tableBucket, INITIAL_OFFSET));
+ bucketGroup.gauge(CURRENT_OFFSET_METRIC_GAUGE, () ->
bucketMetrics.currentOffset());
+ bucketGroup.gauge(LOG_END_OFFSET_METRIC_GAUGE, () ->
bucketMetrics.logEndOffset());
+ bucketGroup.gauge(RECORDS_LAG_METRIC_GAUGE, () ->
bucketMetrics.recordsLag());
+ }
+
+ private long recordsLagMax() {
+ long maxLag = 0L;
+ for (BucketMetrics metrics : bucketMetrics.values()) {
+ maxLag = Math.max(maxLag, metrics.recordsLag());
+ }
+ return maxLag;
+ }
+
+ private long recordsLagSum() {
+ long lagSum = 0L;
+ for (BucketMetrics metrics : bucketMetrics.values()) {
+ lagSum += metrics.recordsLag();
+ }
+ return lagSum;
}
- private void checkTableBucketTracked(TableBucket tableBucket) {
- if (!offsets.containsKey(tableBucket)) {
- LOG.warn("Offset metrics of TableBucket {} is not tracked",
tableBucket);
+ private int laggingBuckets() {
+ int laggingBuckets = 0;
+ for (BucketMetrics metrics : bucketMetrics.values()) {
+ if (metrics.recordsLag() > 0) {
+ laggingBuckets++;
+ }
}
+ return laggingBuckets;
}
public SourceReaderMetricGroup getSourceReaderMetricGroup() {
return sourceReaderMetricGroup;
}
+
+ private static class BucketMetrics {
+ @Nullable private volatile String partitionName;
+ private volatile boolean active;
+ private volatile long currentOffset;
+ private volatile long nextConsumedOffset;
+ private volatile long logEndOffset;
+ private volatile boolean refreshLogEndOffset;
+
+ private BucketMetrics(
+ @Nullable String partitionName,
+ long startingOffset,
+ @Nullable Long stoppingOffset,
+ boolean refreshLogEndOffset) {
+ this.partitionName = partitionName;
+ this.currentOffset = INITIAL_OFFSET;
+ this.nextConsumedOffset = startingOffset;
+ this.logEndOffset = stoppingOffset == null ? UNINITIALIZED :
stoppingOffset;
+ this.refreshLogEndOffset = refreshLogEndOffset && stoppingOffset
== null;
+ this.active = true;
+ }
+
+ private void recordCurrentOffset(long offset) {
+ currentOffset = offset;
+ nextConsumedOffset = offset + 1;
+ }
+
+ private void updateLogEndOffsetIfAdvanced(long logEndOffset) {
+ if (active && refreshLogEndOffset && logEndOffset >
this.logEndOffset) {
+ this.logEndOffset = logEndOffset;
+ }
+ }
+
+ private void deactivate() {
+ active = false;
+ }
+
+ private boolean needsLogEndOffsetRefresh() {
+ return active && refreshLogEndOffset;
+ }
+
+ private SubscribedBucket toSubscribedBucket(TableBucket tableBucket) {
+ return new SubscribedBucket(tableBucket, partitionName);
+ }
+
+ private long currentOffset() {
+ return currentOffset;
+ }
+
+ private long logEndOffset() {
+ return logEndOffset;
+ }
+
+ private long recordsLag() {
+ if (!active || logEndOffset < 0 || nextConsumedOffset < 0) {
+ return 0L;
+ }
Review Comment:
`recordsLag()` currently returns `0` when `logEndOffset` is `UNINITIALIZED`
(< 0). That makes an unknown backlog indistinguishable from “fully caught up”,
which can be misleading right after startup or if offset refresh fails.
Consider returning `UNINITIALIZED` for unknown lag (and teaching the aggregate
gauges to ignore it), or otherwise exposing an explicit “unknown/uninitialized”
state.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java:
##########
@@ -445,6 +480,7 @@ private FlinkRecordsWithSplitIds forLogRecords(ScanRecords
scanRecords) {
List<ScanRecord> bucketScanRecords =
scanRecords.records(scanBucket);
if (!bucketScanRecords.isEmpty()) {
final ScanRecord lastRecord =
bucketScanRecords.get(bucketScanRecords.size() - 1);
+ flinkSourceReaderMetrics.recordLogEndOffset(scanBucket,
lastRecord.logOffset() + 1);
// We keep the maximum message timestamp in the fetch for
calculating lags
Review Comment:
`recordLogEndOffset(scanBucket, lastRecord.logOffset() + 1)` is using the
last consumed record offset (+1) as the *log end offset*. This will
under-report `recordsLag` (often showing 0) whenever the periodic listOffsets
refresh is delayed/disabled, because the “end offset” is being advanced by
consumption rather than by the broker’s latest offset. Consider removing this
update (or only updating log end offset from a real high-watermark/listOffsets
result) so `logEndOffset` truly represents the latest available offset.
--
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]
