[FLINK-40397][mysql] Add binlog position lag metrics for MySQL binlog reader

[hadoopkandy]

What is the purpose of the change

This PR adds binlog position lag metrics for the MySQL binlog reader, which measure the lag between the current consumed binlog offset and the latest master binlog offset.

Unlike the existing currentFetchEventTimeLag metric (which only updates when there are events flowing), these metrics are meaningful even during idle periods — they reflect how far behind the reader is from the MySQL master
at the position level.

The feature is opt-in via scan.binlog.position-lag.interval.ms. A value of -1 (default) disables the feature entirely. When set to a positive value, the connector periodically executes SHOW MASTER STATUS at this
interval to calculate lag.

Brief change log

• Added BinlogLagCalculator to compute binlog position lag, producing two independent results: transaction lag (GTID-based) and byte position lag (file+position based).
• In BinlogSplitReader, conditionally fetch the master's current binlog offset via SHOW MASTER STATUS at the configured interval and store it in a shared AtomicReference<BinlogOffset>.
• In MySqlRecordEmitter, read the shared master offset and calculate lag against the current consumed offset, then report it via MySqlSourceReaderMetrics.
• Registered two new gauge metrics in MySqlSourceReaderMetrics:
  • currentBinlogTransactionLag: GTID-based transaction count lag (-1 when GTID is unavailable)
  • currentBinlogBytePositionLag: byte-level position lag (-1 when position info is unavailable)
• Added configuration option scan.binlog.position-lag.interval.ms (default -1, disabled) to control the feature and polling frequency.

Lag calculation strategy

Mode	Metric	Lag meaning	Calculation
GTID	currentBinlogTransactionLag	Transaction count difference	Sum of (master max txn ID - current max txn ID) per server UUID
File-position (same file)	currentBinlogBytePositionLag	Byte offset difference	master position - current position
File-position (cross file)	currentBinlogBytePositionLag	Estimated byte difference	file sequence diff × 1,000,000 + master position - current position

Verifying this change

This change added tests:

• BinlogLagCalculatorTest: unit tests covering GTID mode (single/multiple UUIDs, disjoint intervals, starts-from-middle, same GTID fallback to position), file-position mode (same file, cross file, edge cases).
• MySqlRecordEmitterTest#testBinlogPositionLagMetricIsUpdated: verifies that the byte position lag metric is updated correctly during record emission through the public API path.

Does this pull request potentially affect one of the following parts

• Dependencies: no
• The public API: yes (new builder method binlogPositionLagIntervalMs and config option)
• The serializers: no
• The runtime per-record code path: yes (lightweight gauge update during binlog phase only, disabled by default)
• Anything that affects determine of shard: no

Documentation

• Does this pull request introduce a new feature? yes — new monitoring metrics and configuration option
• If yes, how is the feature documented? JavaDoc on metric constants; configuration option documented in both English and Chinese user-facing docs (mysql-cdc.md).

[Copilot]

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Adds a new MySQL source metric (currentBinlogPositionLag) to report position-level lag between consumed binlog offset and the master’s latest offset, including during idle periods.

Changes:

• Introduced BinlogLagCalculator (GTID + file/pos modes) and added unit tests.
• Wired a shared AtomicReference<BinlogOffset> from BinlogSplitReader (periodic master offset fetch) to MySqlRecordEmitter (periodic lag reporting).
• Registered a new gauge metric in MySqlSourceReaderMetrics and updated reader/emitter constructors and tests accordingly.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/BinlogLagCalculatorTest.java	Adds unit coverage for GTID and file-position lag calculations.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java	Updates test helpers to pass shared master-offset reference through constructors.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitterTest.java	Adds a test for the new lag metric and updates emitter construction.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java	Updates binlog reader tests to pass the shared master-offset reference.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/BinlogLagCalculator.java	Implements lag computation for GTID and file-position modes.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSplitReader.java	Threads the shared master-offset reference into BinlogSplitReader.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java	Periodically computes and records the new lag metric during binlog phase.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java	Registers currentBinlogPositionLag gauge and stores the latest reported value.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java	Wires shared master-offset reference between split reader and record emitter.
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java	Periodically fetches master binlog offset and exposes it via shared reference.
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java	Updates pipeline emitter constructor to pass shared master-offset reference.
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSource.java	Updates record emitter supplier wiring for the new constructor signature.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[lvyanquan]

Could we first clarify the unit and semantics of currentBinlogPositionLag?

Currently, the same gauge represents different quantities:

• GTID sets differ: the value is an estimated transaction count.
• GTID sets are equal or GTID is disabled: the value is a byte-position difference.
• Binlog files differ: the value is a synthetic estimate based on fileSequenceDiff * 1_000_000.

As a result, the metric may switch units at runtime and cannot be interpreted consistently or used with a stable alert threshold. The cross-file value also has no physical unit.

I think we should define whether this metric represents transaction count, byte lag, or an abstract position distance before finalizing the calculation. If both transaction and byte lag are useful, exposing separate metrics may be clearer.

[lvyanquan]

And Could we make this metric opt-in through a source configuration instead of enabling it by default?

This is not a purely local metric. It executes SHOW MASTER STATUS against the MySQL server every 10 seconds. Enabling it by default means that all existing CDC jobs will start issuing additional queries after upgrading, even when users do not need this metric. When many CDC jobs connect to the same MySQL instance, these recurring queries may accumulate and introduce unexpected server-side overhead.

I suggest disabling this feature by default and providing a configuration option to enable it explicitly. When disabled, the connector should skip both metric registration and the periodic master-status query.

[hadoopkandy]

And Could we make this metric opt-in through a source configuration instead of enabling it by default?

This is not a purely local metric. It executes SHOW MASTER STATUS against the MySQL server every 10 seconds. Enabling it by default means that all existing CDC jobs will start issuing additional queries after upgrading, even when users do not need this metric. When many CDC jobs connect to the same MySQL instance, these recurring queries may accumulate and introduce unexpected server-side overhead.

I suggest disabling this feature by default and providing a configuration option to enable it explicitly. When disabled, the connector should skip both metric registration and the periodic master-status query.

Great feedback. Addressed both concerns:

1. Opt-in via configuration: Added scan.binlog.position-lag.interval.ms (default -1, meaning disabled). A positive value enables the feature and controls the polling interval. When disabled, both the periodic SHOW MASTER
   STATUS query and metric registration are skipped entirely.
2. Metric semantics split: Replaced the single currentBinlogPositionLag with two independent metrics:
   - currentBinlogTransactionLag: GTID-based transaction count lag (available only in GTID mode)
   - currentBinlogBytePositionLag: byte-level lag based on binlog file + position (always available when enabled)

This avoids the ambiguity of mixing different units in one metric and gives users clear, actionable signals.