[
https://issues.apache.org/jira/browse/FLINK-40397?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
wangkang updated FLINK-40397:
-----------------------------
Attachment: image-2026-08-16-16-27-55-610.png
> Add currentBinlogPositionLag metric for MySQL binlog reader
> ------------------------------------------------------------
>
> Key: FLINK-40397
> URL: https://issues.apache.org/jira/browse/FLINK-40397
> Project: Flink
> Issue Type: Improvement
> Components: Flink CDC
> Reporter: wangkang
> Priority: Major
> Attachments: image-2026-08-16-16-27-55-610.png
>
>
> Add a new gauge metric {{currentBinlogPositionLag}} to the MySQL CDC
> connector that measures the binlog position gap between the reader's current
> consumed offset and the MySQL master's latest offset, enabling accurate
> detection of whether the reader has caught up.
> h2. Problem
> The existing {{currentFetchEventTimeLag}} metric is computed as
> {{System.currentTimeMillis() - messageTimestamp}} and only updates when a
> record is processed. This creates a blind spot:
> * If the reader processes a record from 1:00 AM at 10:00 AM, the metric
> reports 9h lag.
> * If no new binlog events are produced after that, the reader is fully
> caught up, yet the metric remains frozen at 9h indefinitely.
> *Impact:*
> * False positives in alerting (metric shows high lag when reader is actually
> idle and current)
> * No way to distinguish "caught up and idle" from "falling behind"
> h2. Solution
> Introduce {{currentBinlogPositionLag}} — a position-based metric that
> compares the reader's current binlog offset against the MySQL master's latest
> offset.
> ||Metric value||Meaning||
> |0|Fully caught up with master|
> |> 0|Unconsumed binlog data exists (higher = further behind)|
> |-1|Still in snapshot phase (binlog reading has not started)|
> h3. GTID Mode
> For each server UUID present in the master's GTID set, compute:
>
> {{lag += max(0, master_max_transaction_id - current_max_transaction_id)}}
> Sum across all UUIDs.
> {quote}Note: We cannot use {{GtidSet.subtract()}} because CDC may resume from
> a checkpoint midpoint (e.g., {{{}uuid:1774595494-1775564172{}}}). subtract()
> would incorrectly count all transactions before the checkpoint start as
> lag.{quote}
> h3. Non-GTID Mode
> * Same binlog file: {{master.position - current.position}} (byte difference)
> * Cross-file: {{(master_file_seq - current_file_seq) × 1_000_000 +
> master.position}} (synthetic weight to reflect cross-file severity; not
> actual byte count)
> h2. Design
>
> {{MySqlSource.createReader()
> |
> |--> creates AtomicReference<BinlogOffset> (shared bridge)
> | | |
> | v v
> | MySqlSplitReader MySqlRecordEmitter
> | |
> | v
> | BinlogSplitReader
> |
> |
> | Fetcher thread (BinlogSplitReader.pollSplitRecords):
> | - Every ~10s: SHOW MASTER STATUS
> | - Write result to AtomicReference
> |
> | Main thread (RecordEmitter.processElement):
> | - On data change record or heartbeat event:
> | - Read AtomicReference
> | - Compute lag
> | - Report metric: currentBinlogPositionLag}}
> h3. Overhead
> ||Dimension||Impact||
> |MySQL load|{{SHOW MASTER STATUS}} reads from memory, ~μs latency, once per
> 10s|
> |Connections|None — reuses existing JDBC connection|
> |Threads|None — piggybacks on fetcher thread's poll loop|
> |Data path|None — lag calculation happens after record emission|
> |Memory|One AtomicReference (~16 bytes)|
> h2. Alternatives Considered
> ||Approach||Rejected because||
> |Independent scheduled thread|Extra thread + JDBC connection; lifecycle
> complexity not justified for a metric|
> |Heartbeat-driven only|Depends on user enabling
> {{{}heartbeat.interval.ms{}}}; heartbeat offset doesn't advance without new
> binlog events|
> |Enumerator-side broadcast|Requires custom SourceEvent communication; high
> implementation cost; offset alignment across readers is complex|
> h2. Compatibility
> * No breaking changes to public APIs
> * Purely additive (new metric alongside existing ones)
> * Works with GTID and non-GTID MySQL configurations
> * Compatible with MySQL 5.7, 8.0, and 8.4+
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
