How We Fixed Flink CDC GC Crashes and Cut Latency from 30min to 5s

Scenario & Problems: From Data Latency to Task Crashes

The core data pipeline: 20 MySQL instances (different business databases) → Flink CDC (incremental capture) → Kafka (buffering & peak-shaving) → Flink (multi-stream Join / cleansing) → StarRocks (real-time warehouse storage) . Requirement: end-to-end latency ≤ 60 seconds.

Core Failure Symptoms

Resource bottleneck: TaskManager only 1C/2G, parallelism=1, single node handling binlog read/parse/write for 20 MySQL instances — severe overload.

GC stalls: JVM GC abnormal, 831 ms/s GC time over 3 minutes, 83% CPU spent on GC, core processing completely stalled. (TaskManager GC time severe, longest per-second GC time [831.29 ms/s] in PT3M)

Checkpoint failures: Insufficient memory caused state persistence timeout, checkpoint success rate only 97%, frequent restarts worsened backlog.

Uncontrolled latency: CDC production 2000 records/s, downstream processing only 200 records/s, backlog grew continuously, StarRocks latency > 30 minutes.

Startup strategy risk: Used StartupOptions.latest() — reads only from latest binlog position; if binlog cleaned during restart (via expire_logs_days), incremental data lost permanently.

Root Cause Analysis: From Symptoms to Essence

1. Severe Resource Misconfiguration

TaskManager memory monitoring (Figure 1) revealed three imbalances:

Framework Heap usage 90.31% — Flink framework memory nearly exhausted.

Task Heap over-allocated: configured 538 MB, actual usage 599 MB — heap OOM risk.

Managed Memory idle waste: 635 MB unused, starving state caching & serialization.

2. Parallelism Design Mismatched to Load

Parallelism=1 forced all 20 MySQL binlog reading, parsing, Kafka writing into a single slot:

IO-intensive ops blocked: single-threaded concurrent handling of 20 binlog streams, network/disk IO became core bottleneck.

No parallel processing capability: couldn't leverage cluster distributed resources, production speed far exceeded processing speed, backlog propagated downstream.

3. Hidden Risks of StartupOptions.latest()

Two fatal defects:

Data loss risk: If task stops and MySQL binlog cleaned by expire_logs_days, restart cannot read missed increments — data integrity broken.

Fails warehouse full+incremental need: First startup only syncs post-start increments, missing historical full data, affecting downstream analysis accuracy.

4. Missing Parameter Configurations Amplify Failures

Debezium batch read params not configured — single-record processing, frequent network round-trips, low read efficiency.

Checkpoint policy unreasonable: timeout too short (default 1 min), memory pressure causes state persistence timeout → checkpoint failures.

Heartbeat mechanism not enabled — long-idle MySQL connections drop, reconnection worsens latency.

System Optimization: From Emergency Recovery to Root-Cause Fix

Phase 1: Emergency Recovery — 1 Hour to Relieve Latency & Crashes

1. Temporary Resource Scale-Up

Adjusted TaskManager from 1C/2G to 4C/32G:

# flink-conf.yaml (CDC task dedicated config)
taskmanager.memory.process.size: 32G  # process total memory
taskmanager.numberOfTaskSlots: 4      # slots per TaskManager
taskmanager.memory.task.heap.size: 4G # task heap memory
taskmanager.memory.framework.heap.size: 512MB # framework heap memory

2. Split Tasks + Increase Parallelism

Grouped 20 MySQL instances by business domain into 5 sub-tasks, each parallelism=4, total parallelism=20, achieving load balance:

// Group by business domain, split 20 MySQL instances into 5 groups
List<String> dbGroups = Arrays.asList(
    "db_order1,db_order2,db_order3,db_order4",  // Order domain
    "db_user1,db_user2,db_user3,db_user4",      // User domain
    "db_pay1,db_pay2,db_pay3,db_pay4",          // Payment domain
    "db_goods1,db_goods2,db_goods3,db_goods4",  // Goods domain
    "db_oper1,db_oper2,db_oper3,db_oper4"        // Operations domain
);
// Each group builds independent CDC Source, parallelism=4
for (String dbGroup : dbGroups) {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("mysql-cluster-host")
        .databaseList(dbGroup.split(","))  // each group 4 databases
        .username("cdc_sync_user")
        .password("xxx@2024")
        .startupOptions(StartupOptions.latest())  // temporary for emergency, optimize later
        .deserializer(new StringDebeziumDeserializationSchema())  // custom deserializer
        .build();
    // Launch CDC task, parallelism=4
    env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "CDC-Source-" + dbGroup)
        .setParallelism(4)
        .sinkTo(kafkaSink);  // write to corresponding Kafka topic
}

Fix note: Real-time dashboard used mainly daytime. Because StartupOptions.latest() was used, restarting after resource increase would cause data inaccuracy. To ensure accuracy, restart performed after 11 PM (this is a pitfall: slow vs. data accuracy trade-off, to be optimized later).

Phase 2: Root-Cause Fix — Solve GC, Checkpoint & Startup Strategy

1. Optimize Memory Model & GC Parameters

Enabled Managed Memory, allocated regions reasonably, configured G1GC to reduce Full GC frequency:

# Memory model optimization
taskmanager.memory.framework.heap.size: 512MB  # framework heap expanded
taskmanager.memory.managed.size: 2G            # enable 2G managed memory for state cache, serialization
taskmanager.memory.network.min: 512MB          # network buffer min
taskmanager.memory.network.max: 1G             # network buffer max

Optimization logic: G1GC via region division & concurrent collection controls single GC pause ≤ 200 ms; 45% heap occupancy triggers collection, avoids OOM; parallel reference processing improves collection efficiency.

2. Optimize Checkpoint Strategy

Enabled unaligned checkpoints, extended timeout, adapted to memory-constrained state persistence:

# Checkpoint core config
execution.checkpointing.interval: 30s          # checkpoint interval 30s
execution.checkpointing.timeout: 10min         # timeout extended to 10 minutes
execution.checkpointing.unaligned-checkpoints.enabled: true  # enable unaligned checkpoints
execution.checkpointing.mode: EXACTLY_ONCE     # exactly-once semantics
state.backend: rocksdb                         # RocksDB state backend for large state
state.checkpoints.dir: hdfs:///flink/checkpoints/cdc-task  # checkpoint storage (HDFS)

3. Fix Startup Strategy to Avoid Data Loss

Replaced StartupOptions.latest() with StartupOptions.initial(), combined with incremental snapshot for integrated full+incremental sync:

// Optimized CDC Source build logic
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
    .hostname("mysql-cluster-host")
    .databaseList(dbGroup.split(","))
    .username("cdc_sync_user")
    .password("xxx@2024")
    .startupOptions(StartupOptions.initial())  // initial strategy: full snapshot + incremental binlog
    .snapshotMode(SnapshotMode.INCREMENTAL)    // enable incremental snapshot, avoid full table scan
    .deserializer(new StringDebeziumDeserializationSchema())
    .build();

Optimization: First startup executes incremental snapshot (scans only new data, avoids full-table scan cost), then continuously reads binlog increments; combined with Checkpoint & RocksDB state backend, restart recovers from latest checkpoint without re-running full snapshot.

Supplementary note: Above are optional optimization choices; actual case was resource insufficiency plus startup strategy issues.

Risk Governance: StartupOptions Selection & Core Parameter Tuning

1. StartupOptions Full-Scenario Selection Guide

Different startup strategies fit different scenarios; choose based on data integrity needs & startup efficiency. Comparison table below:

Production recommendation: Real-time warehouse prioritizes StartupOptions.initial() with SnapshotMode.INCREMENTAL incremental snapshot, balancing data integrity & startup efficiency.

2. Flink CDC Core Parameter Best Configuration

Optimized core parameters for MySQL CDC sync stability & efficiency:

Best Practices: From Stable Operation to Long-Term Assurance

1. Fault Recovery & Data Safety

State backend selection: Production prefers RocksDB state backend, supports large state persistence, with HDFS checkpoint storage to avoid state loss on restart.

Binlog backup fallback: MySQL set expire_logs_days=7 to extend retention; periodically backup binlog to object storage (S3, OSS); extreme loss recoverable via binlog replay.

Deduplication & idempotency: Use binlog ts_ms (timestamp) + business primary key for idempotent processing, avoid duplicate writes to StarRocks on restart.

Kafka write restart deduplication: Combine Flink state management with Kafka message features — two core approaches: (1) Flink state records written Kafka binlog positions; (2) Kafka message Key idempotent writes.

2. Monitoring & Alerting System

Build full-chain monitoring metrics for early warning:

Resource monitoring: TaskManager CPU > 80% alert, memory > 85% alert.

GC monitoring: Per-second GC time > 200 ms alert, Full GC frequency > 3/hour alert.

Checkpoint monitoring: Success rate < 100% alert, duration > 5 min alert.

Data latency monitoring: CDC source to Kafka consumer > 30s alert, StarRocks end-to-end > 10s alert.

Connection monitoring: MySQL CDC connection count, Kafka producer/consumer connection status (immediate alert on abnormal disconnect).

Optimization Results Verification

After implementing above optimizations, task metrics improved significantly:

GC metrics: Per-second GC time dropped from 831 ms to ≤ 150 ms, CPU GC share < 15%.

Checkpoint metrics: Success rate from 97% to 100%, no timeouts/failures.

Data latency: End-to-end from 30 minutes to ≤ 5 seconds, meets real-time warehouse SLA.

Resource utilization: TaskManager CPU stable 50%-70%, memory ≤ 80%, balanced load.

Stability: 72-hour continuous run without restart, 100% data integrity (no loss, no duplicates).

Summary & Outlook

Stable Flink CDC operation in real-time warehouse relies on coordinated optimization across four dimensions: resource configuration, parallelism design, startup strategy, parameter tuning . Resources must match task load, avoid "small horse pulling big cart"; parallelism must be scientifically designed per data source count & cluster scale to unleash distributed processing; startup strategy must prioritize data integrity — real-time warehouse prefers initial() + incremental snapshot; long-term assurance depends on monitoring/alerting, state persistence, binlog backup mechanisms to safeguard against failures.

This optimization plan validated in "20 MySQL → StarRocks" real scenario, directly reusable for similar CDC sync cases. Follow-up will share remaining pitfall-fixing practices mentioned in article.