[
https://issues.apache.org/jira/browse/FLINK-40397?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
wangkang updated FLINK-40397:
-----------------------------
Description:
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() │
│ │
│ AtomicReference<BinlogOffset> (shared bridge) │
│ ↓ ↓ │
│ MySqlSplitReader MySqlRecordEmitter │
│ ↓ │
│ BinlogSplitReader │
└─────────────────────────────────────────────────────────┘
Fetcher thread (BinlogSplitReader.pollSplitRecords):
→ Every ~10s: SHOW MASTER STATUS → write to AtomicReference
Main thread (RecordEmitter.processElement):
→ On data change record or heartbeat event:
→ Read AtomicReference, compute lag, report metric}}
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+
was:
## Summary
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.
## 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"
## 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) |
### 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.
> 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.
### 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)
## Design
┌─────────────────────────────────────────────────────────┐
│ MySqlSource.createReader() │
│ │
│ AtomicReference (shared bridge) │
│ ↓ ↓ │
│ MySqlSplitReader MySqlRecordEmitter │
│ ↓ │
│ BinlogSplitReader │
└─────────────────────────────────────────────────────────┘
Fetcher thread (BinlogSplitReader.pollSplitRecords):
→ Every ~10s: SHOW MASTER STATUS → write to AtomicReference
Main thread (RecordEmitter.processElement):
→ On data change record or heartbeat event:
→ Read AtomicReference, compute lag, report metric
### 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) |
## 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 |
## 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+
---
> 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
>
> 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() │
> │ │
> │ AtomicReference<BinlogOffset> (shared bridge) │
> │ ↓ ↓ │
> │ MySqlSplitReader MySqlRecordEmitter │
> │ ↓ │
> │ BinlogSplitReader │
> └─────────────────────────────────────────────────────────┘
> Fetcher thread (BinlogSplitReader.pollSplitRecords):
> → Every ~10s: SHOW MASTER STATUS → write to AtomicReference
> Main thread (RecordEmitter.processElement):
> → On data change record or heartbeat event:
> → Read AtomicReference, compute lag, report metric}}
> 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)
