Copilot commented on code in PR #4509:
URL: https://github.com/apache/flink-cdc/pull/4509#discussion_r3801995484
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java:
##########
@@ -151,6 +162,31 @@ private void reportMetrics(SourceRecord element) {
}
}
+ private void reportBinlogLag(MySqlSplitState splitState) {
+ if (!splitState.isBinlogSplitState()) {
+ return;
+ }
+ long now = System.currentTimeMillis();
+ if (now - lastReportBinlogLagTime < REPORT_BINLOG_LAG_INTERVAL_MS) {
+ return;
+ }
+ lastReportBinlogLagTime = now;
+ BinlogOffset currentOffset =
splitState.asBinlogSplitState().getStartingOffset();
+ BinlogOffset masterOffset = latestMasterOffset.get();
+ if (currentOffset == null || masterOffset == null) {
+ return;
+ }
Review Comment:
`lastReportBinlogLagTime` is updated before validating that both offsets are
available. If either offset is null, the method returns but suppresses
reporting for 10s, delaying the first successful metric update. Move the
timestamp update to after the null-check (or after a successful calculation) so
missing offsets don’t throttle subsequent attempts.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java:
##########
@@ -219,6 +234,20 @@ public Iterator<SourceRecords> pollSplitRecords() throws
InterruptedException {
sourceRecords.add(event.getRecord());
}
}
+
+ // Fetch master offset AFTER polling records, so master reflects
the latest
+ // state and is always >= the records we just polled.
+ long now = System.currentTimeMillis();
+ if (now - lastFetchMasterStatusTime >
FETCH_MASTER_STATUS_INTERVAL_MS) {
+ try {
+ latestMasterOffset.set(
+
currentBinlogOffset(statefulTaskContext.getConnection()));
+ lastFetchMasterStatusTime = now;
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch master binlog offset for lag
metric", e);
Review Comment:
On repeated failures, `lastFetchMasterStatusTime` is never updated, so this
block can run on every `pollSplitRecords()` call after the interval elapses,
potentially creating a high-volume warn log storm. Consider updating a
`lastFetchAttemptTime` (or updating `lastFetchMasterStatusTime` in a `finally`)
and logging with throttling/backoff to keep failure behavior bounded.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java:
##########
@@ -25,6 +25,7 @@
/** A collection class for handling metrics in {@link MySqlSourceReader}. */
public class MySqlSourceReaderMetrics {
+ public static final String CURRENT_BINLOG_POSITION_LAG =
"currentBinlogPositionLag";
Review Comment:
Metric names appear to be centralized in `MetricNames` (as used for
`CURRENT_FETCH_EVENT_TIME_LAG`), but the new metric name is introduced as a
constant in `MySqlSourceReaderMetrics`. To keep naming consistent and avoid
divergent sources of truth, consider adding `CURRENT_BINLOG_POSITION_LAG` to
`MetricNames` and referencing it from here.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitterTest.java:
##########
@@ -94,7 +97,44 @@ record -> {
.isEqualByComparingTo(fakeOffset);
}
+ @Test
+ void testBinlogPositionLagMetricIsUpdated() throws Exception {
+ MySqlSourceReaderMetrics metrics =
+ new MySqlSourceReaderMetrics(
+
UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup());
+ metrics.registerMetrics();
+
+ AtomicReference<BinlogOffset> masterOffset =
+ new
AtomicReference<>(BinlogOffset.ofBinlogFilePosition("mysql-bin.000001", 10000));
+
+ MySqlRecordEmitter<Void> emitter = createRecordEmitter(metrics,
masterOffset);
+
+ MySqlBinlogSplitState splitState = createBinlogSplitState();
+
splitState.setStartingOffset(BinlogOffset.ofBinlogFilePosition("mysql-bin.000001",
3000));
+
+ Method reportMethod =
+ MySqlRecordEmitter.class.getDeclaredMethod(
+ "reportBinlogLag", MySqlSplitState.class);
+ reportMethod.setAccessible(true);
+ reportMethod.invoke(emitter, splitState);
Review Comment:
This test uses reflection to invoke a private method, which is brittle
(method name/signature changes will silently break the test). Prefer exercising
the behavior through the public `processElement(...)` path, or make
`reportBinlogLag` package-private and annotate it with `@VisibleForTesting` so
the test can call it directly without reflection.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java:
##########
@@ -35,13 +36,21 @@ public class MySqlSourceReaderMetrics {
*/
private volatile long fetchDelay = UNDEFINED;
+ /**
+ * The binlog position lag between current consumed offset and the latest
master offset. This
+ * metric is meaningful even when there is no data flowing, as it reflects
how far behind the
+ * reader is from the MySQL master.
+ */
+ private volatile long binlogPositionLag = UNDEFINED;
+
public MySqlSourceReaderMetrics(MetricGroup metricGroup) {
this.metricGroup = metricGroup;
}
public void registerMetrics() {
metricGroup.gauge(
MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, (Gauge<Long>)
this::getFetchDelay);
+ metricGroup.gauge(CURRENT_BINLOG_POSITION_LAG, (Gauge<Long>)
this::getBinlogPositionLag);
Review Comment:
Metric names appear to be centralized in `MetricNames` (as used for
`CURRENT_FETCH_EVENT_TIME_LAG`), but the new metric name is introduced as a
constant in `MySqlSourceReaderMetrics`. To keep naming consistent and avoid
divergent sources of truth, consider adding `CURRENT_BINLOG_POSITION_LAG` to
`MetricNames` and referencing it from here.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/BinlogLagCalculator.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.mysql.source.utils;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
+import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+
+import io.debezium.connector.mysql.GtidSet;
+
+/**
+ * Calculator for binlog position lag between the current consumed offset and
the latest master
+ * offset. Supports both GTID mode and file-position mode.
+ */
+@Internal
+public class BinlogLagCalculator {
+
+ /** Cached parsed master GtidSet to avoid re-parsing on every calculation.
*/
+ private String cachedMasterGtidSetStr;
+
+ private GtidSet cachedMasterGtidSet;
+
+ /**
+ * Calculate binlog position lag between current offset and master offset.
+ *
+ * @param current the current consumed binlog offset
+ * @param master the latest master binlog offset
+ * @return the lag value, or -1 if cannot determine
+ */
+ public long calculateLag(BinlogOffset current, BinlogOffset master) {
+ // GTID mode: compare max transaction ID per UUID
+ String masterGtidSetStr = master.getGtidSet();
+ String currentGtidSetStr = current.getGtidSet();
+ if (masterGtidSetStr != null
+ && !masterGtidSetStr.isEmpty()
+ && currentGtidSetStr != null
+ && !currentGtidSetStr.isEmpty()
+ && !masterGtidSetStr.equals(currentGtidSetStr)) {
+ // Cache parsed master GtidSet (only changes every 10s)
+ if (!masterGtidSetStr.equals(cachedMasterGtidSetStr)) {
+ cachedMasterGtidSet = new GtidSet(masterGtidSetStr);
+ cachedMasterGtidSetStr = masterGtidSetStr;
+ }
+ GtidSet masterSet = cachedMasterGtidSet;
+ GtidSet currentSet = new GtidSet(currentGtidSetStr);
+ // Calculate lag as the difference between max transaction IDs.
+ // We cannot use subtract() because the current GTID set may not
start from 1
+ // (CDC starts from a checkpoint midpoint), which would
incorrectly count
+ // historical transactions as lag.
+ long lag = 0;
+ for (GtidSet.UUIDSet masterUuidSet : masterSet.getUUIDSets()) {
+ long masterMax = getMaxTransactionId(masterUuidSet);
+ GtidSet.UUIDSet currentUuidSet =
+ currentSet.forServerWithId(masterUuidSet.getUUID());
+ if (currentUuidSet == null) {
+ // Current has no transactions for this server UUID, count
all of master's
+ lag += masterMax;
+ } else {
+ long currentMax = getMaxTransactionId(currentUuidSet);
+ lag += Math.max(0, masterMax - currentMax);
+ }
+ }
+ return lag;
+ }
+
+ // Non-GTID mode: compare file + position
+ String masterFile = master.getFilename();
+ String currentFile = current.getFilename();
+ if (masterFile != null && masterFile.equals(currentFile)) {
+ return Math.max(0, master.getPosition() - current.getPosition());
+ }
+ // Cross-file: compare file sequence number
+ if (masterFile != null && currentFile != null) {
+ try {
+ long masterSeq = extractFileSequence(masterFile);
+ long currentSeq = extractFileSequence(currentFile);
+ if (masterSeq > currentSeq) {
+ // Rough estimate: actual file size is unknown, so we use
1,000,000 as a
+ // synthetic weight per file gap to produce a
monotonically increasing lag
+ // value that indicates cross-file distance. This is NOT
actual byte lag.
+ return (masterSeq - currentSeq) * 1_000_000L +
master.getPosition();
Review Comment:
For cross-file comparisons, the returned estimate does not incorporate
`current.getPosition()`. That means the lag won’t decrease as the reader
advances within the current (older) file, which makes the metric
misleading/non-responsive during normal catch-up. Include the current position
in the estimate (and clamp at 0) so progress within a file reduces the reported
lag.
--
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]
