[FLINK-40397][mysql] Add currentBinlogPositionLag metric for MySQL binlog reader - #4509
Open
hadoopkandy wants to merge 1 commit into
Open
[FLINK-40397][mysql] Add currentBinlogPositionLag metric for MySQL binlog reader#4509hadoopkandy wants to merge 1 commit into
hadoopkandy wants to merge 1 commit into
Conversation
Contributor
There was a problem hiding this comment.
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>fromBinlogSplitReader(periodic master offset fetch) toMySqlRecordEmitter(periodic lag reporting). - Registered a new gauge metric in
MySqlSourceReaderMetricsand 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.
Comment on lines
+173
to
+178
| lastReportBinlogLagTime = now; | ||
| BinlogOffset currentOffset = splitState.asBinlogSplitState().getStartingOffset(); | ||
| BinlogOffset masterOffset = latestMasterOffset.get(); | ||
| if (currentOffset == null || masterOffset == null) { | ||
| return; | ||
| } |
Comment on lines
+245
to
+247
| lastFetchMasterStatusTime = now; | ||
| } catch (Exception e) { | ||
| LOG.warn("Failed to fetch master binlog offset for lag metric", e); |
| // 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(); |
Comment on lines
+115
to
+119
| Method reportMethod = | ||
| MySqlRecordEmitter.class.getDeclaredMethod( | ||
| "reportBinlogLag", MySqlSplitState.class); | ||
| reportMethod.setAccessible(true); | ||
| reportMethod.invoke(emitter, splitState); |
| /** A collection class for handling metrics in {@link MySqlSourceReader}. */ | ||
| public class MySqlSourceReaderMetrics { | ||
|
|
||
| public static final String CURRENT_BINLOG_POSITION_LAG = "currentBinlogPositionLag"; |
Comment on lines
51
to
+53
| metricGroup.gauge( | ||
| MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, (Gauge<Long>) this::getFetchDelay); | ||
| metricGroup.gauge(CURRENT_BINLOG_POSITION_LAG, (Gauge<Long>) this::getBinlogPositionLag); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
This PR adds a new metric
currentBinlogPositionLagfor the MySQL binlog reader, which measures the lag between the current consumed binlog offset and the latest masterbinlog offset.
Unlike the existing
currentFetchEventTimeLagmetric (which only updates when there are events flowing), this metric is meaningful even during idle periods — it reflects howfar behind the reader is from the MySQL master at the position level.
Brief change log
BinlogLagCalculatorto compute binlog position lag, supporting both GTID mode and file-position mode.BinlogSplitReader, periodically (every 10s) fetch the master's current binlog offset viaSHOW MASTER STATUSand store it in a sharedAtomicReference<BinlogOffset>.MySqlRecordEmitter, periodically (every 10s) read the shared master offset and calculate the lag against the current consumed offset, then report it viaMySqlSourceReaderMetrics.currentBinlogPositionLaginMySqlSourceReaderMetrics.Lag calculation strategy
Verifying this change
This change added tests:
BinlogLagCalculatorTest: unit tests covering GTID mode (single/multiple UUIDs, disjoint intervals, starts-from-middle), file-position mode (same file, cross file, edgecases).
MySqlRecordEmitterTest#testBinlogPositionLagMetricIsUpdated: verifies that the lag metric is updated correctly during record emission.Does this pull request potentially affect one of the following parts
Documentation