Flink CDC Connectors: Schema-First Streaming Ingest at Scale

flink cdc is the ingest stack that quietly replaced Debezium + Kafka + a downstream Flink job on every serious 2026 lakehouse deployment — a single Apache Flink job that reads change events from Postgres, MySQL, MongoDB, or Oracle and writes them straight into Doris, StarRocks, Paimon, Snowflake, BigQuery, or Iceberg with no Kafka hop in the middle. The pre-3.x world stitched three services together: Debezium tailed the WAL / binlog into Kafka, a Flink job consumed Kafka and applied projection logic, and a separate sink writer emitted rows into the analytics store. Three hops, three checkpointing stories, three schema-evolution policies, and three on-call rotations to keep alive. Flink CDC 3.x collapses that pipeline into one Flink graph driven by a 40-line YAML file — and the engineering conversation shifts from how do I glue these services to which source, which sink, which pool of behaviours do I opt into on schema evolution.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "walk me through a flink cdc pipeline for MySQL to Doris" or "how does the flink cdc yaml DSL handle flink cdc schema evolution when a source table adds a column?" or "does Flink CDC give exactly-once, and how does the debezium engine embedded inside flink cdc connector compare to running Debezium standalone?" It walks through why the Debezium + Kafka + Sink stack was ripe for collapse, the source-connector matrix across flink cdc postgres, flink cdc mysql, MongoDB, and Oracle, the pipeline YAML DSL with source-transform-route-sink sections, the schema-evolution behaviour matrix that auto-applies ADD COLUMN into flink cdc snowflake or Paimon or Iceberg targets, and the production patterns — savepoints, EOS, backfill throttling, sink-side idempotency — that senior engineers ship into every ingest deployment. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the streaming practice library →, rehearse on the ETL practice library →, and sharpen the tuning axis with the optimization practice library →.

---

On this page

• Why Flink CDC 3.x collapsed the Debezium + Kafka + Sink stack
• Source connectors — Postgres / MySQL / Mongo / Oracle
• Pipeline YAML — declarative topology
• Schema evolution + at-least-once vs EOS
• Sink connectors + comparison to Debezium
• Cheat sheet — Flink CDC recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why Flink CDC 3.x collapsed the Debezium + Kafka + Sink stack

From three hops to one Flink job — the architectural story that made Flink CDC the 2026 default

The one-sentence invariant: before Flink CDC 3.x, streaming ingest from a Postgres or MySQL source into a Doris or Iceberg sink required three separate services (Debezium reading the log into Kafka, a Flink or Spark job consuming Kafka, and a sink writer emitting rows) — and Flink CDC 3.x collapses that entire pipeline into a single Flink job driven by a YAML file. The architectural collapse is not just an ergonomic upgrade. It removes an entire Kafka cluster from the operational footprint for pipelines that never needed the durability layer, it unifies three checkpointing stories into one, and it lets schema evolution propagate end-to-end without a fan-out of coordination code. Every senior interview conversation about a modern ingest stack begins with the question "which pattern are you on — the legacy three-hop or the new single-job?" and the follow-up "why?"

The four axes interviewers actually probe.

• Source connector. Postgres via pgoutput logical replication, MySQL via binlog + GTID, MongoDB via change streams, Oracle via LogMiner or XStream. Each connector wraps a protocol-specific log tap plus a snapshot phase; the senior signal is knowing the specific replication mechanism for each source (pgoutput for Postgres, GTID-anchored binlog for MySQL, resume tokens for Mongo) and the failure modes they add (replication slot leakage, GTID gaps, oplog window expiry).
• Pipeline YAML. The Flink CDC 3.x DSL exposes four sections — source, transform, route, sink — that together replace 500 lines of Java DataStream code for the majority of ingest pipelines. Knowing when the YAML runs out (custom UDF logic, non-trivial joins) and when to fall back to Flink DataStream or Flink SQL is the senior architectural signal.
• Schema evolution. The killer feature. When the source table runs ALTER TABLE users ADD COLUMN email VARCHAR(255), Flink CDC propagates the DDL through the pipeline and issues the equivalent ALTER TABLE on the sink automatically. The behaviour matrix — lenient (auto-apply, default) vs strict (fail-fast on unknown ops) — is the interview question every senior candidate should answer without prompting.
• Sink connector + EOS. Doris, StarRocks, Paimon, Iceberg, Snowflake, and BigQuery all have first-party Flink CDC sinks. Exactly-once semantics come from Flink checkpointing plus a transactional sink write (two-phase commit or an idempotent-by-primary-key upsert). The senior answer names the specific mechanism per sink — stream_load transactions for Doris, transactional writes for Iceberg via the catalog commit protocol, MERGE-into-idempotent-target for Snowflake.

Why the legacy three-hop stack was ripe for collapse.

• Ops surface. Debezium runs as a Kafka Connect worker with its own JVM, its own checkpointing (offset commits to Kafka), and its own failure modes (rebalance storms, offset lag). Kafka itself is an entire service — brokers, controllers, ZooKeeper or KRaft, tiered storage. The downstream Flink job is a third JVM cluster with a fourth checkpointing story. Every hop is a page-able rotation.
• Latency. Each hop adds ~100 ms of buffering. A change event that took 5 ms on the source WAL takes 300–500 ms to land in Doris via Debezium → Kafka → Flink → sink. For interactive analytics dashboards where SLA is "sub-second staleness," the three-hop latency was already at the edge.
• Schema evolution. In the legacy stack, a source ADD COLUMN requires coordinated changes in Debezium (regenerate the Avro/JSON schema in the schema registry), Kafka (the topic schema evolves), the Flink job (the TableSchema deserializer needs the new column), and the sink writer (the target table needs the new column). Four teams, four PRs, one release window. Most teams simply stopped evolving schemas online — which is a design smell.
• Data lineage. Three hops means three points where data can silently drop. Kafka retention expiry, Debezium offset skew, Flink watermark gaps — each is a subtle divergence between source and sink that only surfaces at reconciliation time.
• Cost. Kafka storage, Debezium worker CPU, Flink cluster CPU. A three-hop pipeline runs about 3× the infrastructure of the equivalent Flink CDC 3.x single-job pipeline.

What Flink CDC 3.x actually is.

• Runtime. A regular Apache Flink job. Not a new service. The Flink CDC connectors are Flink source and sink connectors packaged with a YAML-driven job builder.
• Job builder. The flink-cdc-pipeline-connector-* JARs and a YAML parser that reads pipeline.yaml, translates it into a Flink DataStream graph, and submits the job to a Flink cluster. Under the hood it's still Flink — the YAML is a facade over the DataStream API.
• Sources. The flink-cdc-connector-postgres, flink-cdc-connector-mysql, flink-cdc-connector-mongodb, flink-cdc-connector-oracle families. Each wraps the Debezium engine (yes, the same Debezium codebase) as an embedded library, not as a separate Kafka Connect process. The Debezium engine reads the source log; the Flink source connector emits the change events into the Flink pipeline directly.
• Sinks. First-party sinks for Doris, StarRocks, Paimon, and increasingly Iceberg, Snowflake, BigQuery, and Kafka (for teams that still want the durability layer). Each sink implements the Flink SinkV2 API with two-phase-commit semantics wherever the target system supports it.
• YAML DSL. Four top-level sections — source, transform, route, sink — plus a pipeline header for parallelism, checkpoint interval, and name.

What interviewers listen for.

• Do you say "embedded Debezium engine" when asked how Flink CDC reads from Postgres or MySQL? — senior signal.
• Do you mention schema-evolution auto-propagation as the killer feature, not the YAML DSL? — senior signal.
• Do you push back on "we still need Kafka" with the "only if you need multiple downstream consumers or long retention" answer? — required.
• Do you describe Flink CDC as "one Flink job that reads source and writes sink" rather than as "a new tool"? — required.

Worked example — legacy stack line count vs Flink CDC pipeline YAML

Detailed explanation. The clearest way to internalise the collapse is to count the lines of code. The legacy Debezium + Kafka + Flink ingest for a single-table MySQL → Doris pipeline lands at roughly 500 lines split across three services (Debezium connector config, a Flink DataStream job in Java, sink writer glue). The equivalent Flink CDC 3.x pipeline is 40 lines of YAML. The 12× compression is not just cosmetic — every removed line is a piece of state that cannot go wrong at 3 AM.

• Legacy count. Debezium connector JSON (~80 lines), Kafka Connect worker config (~50 lines), Flink Java DataStream job (~250 lines), sink writer (~120 lines).
• Flink CDC count. One YAML file (~40 lines) plus one Flink cluster (~0 net additions since Flink was already in the stack).
• The math is 12×. Not marketing — this is a direct count of files under version control.

Question. A team ships a legacy Debezium + Kafka + Flink pipeline that consumes ~500 lines of code across three services. Quantify the collapse into Flink CDC 3.x and describe what specifically disappears.

Input.

Component (legacy)	Lines / files	Runtime
Debezium connector JSON	80 lines	Kafka Connect worker JVM
Kafka Connect worker.properties	50 lines	separate JVM
Flink DataStream Java job	250 lines	Flink cluster JVM
Sink writer + schema glue	120 lines	inside Flink job
Total	~500 lines, 3 services	3 JVM tiers

Code.

# pipeline.yaml — the entire Flink CDC 3.x replacement for the legacy stack
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: cdc_reader
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.\.*
  server-id: 5400-5404
  server-time-zone: UTC

sink:
  type: doris
  name: orders-doris
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}
  table.create.properties.replication_num: 3
  sink.enable-delete: true

pipeline:
  name: mysql-orders-to-doris
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

// Legacy equivalent — abbreviated Flink DataStream sketch (deleted after migration)
public class LegacyOrdersPipeline {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(30_000);

        // 1. Kafka source with Debezium-produced JSON envelope
        KafkaSource<String> src = KafkaSource.<String>builder()
            .setBootstrapServers("kafka.internal:9092")
            .setTopics("dbserver1.orders.orders")
            .setGroupId("orders-flink")
            .setStartingOffsets(OffsetsInitializer.committedOffsets(
                OffsetResetStrategy.EARLIEST))
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();

        DataStream<String> raw = env.fromSource(src,
            WatermarkStrategy.noWatermarks(), "orders-kafka");

        // 2. Debezium envelope decoder — 60 lines of case-class parsing
        DataStream<OrderRow> parsed = raw.map(new DebeziumEnvelopeParser());

        // 3. Sink writer to Doris with StreamLoad — 120 lines
        parsed.addSink(new DorisSinkFunction("doris-fe.internal:8030",
            "orders_db.orders", ...));

        env.execute("orders-legacy");
    }
}

Step-by-step explanation.

1. The Flink CDC YAML has three sections. source names a MySQL instance, the credentials, and a regex-matched set of tables. sink names a Doris FE (frontend) endpoint and the write credentials. pipeline sets the job name, parallelism, and the schema-change policy. That is the entire pipeline.
2. The legacy sketch above starts a Kafka source, decodes the Debezium JSON envelope (which itself has a nested payload.before / payload.after structure), then routes into a sink function that opens HTTP streams to Doris. Each of these three phases is ~80–120 lines in production.
3. The disappearance list: Kafka bootstrap servers config, Kafka topic naming (Debezium's dbserver1.orders.orders), Kafka consumer group management, Debezium JSON envelope schema, sink batching logic, sink retry logic. All gone. The YAML source is the Debezium engine (embedded); the YAML sink is the StreamLoad transactional writer (embedded).
4. What remains in the YAML is the small set of decisions that actually matter: which source, which credentials, which tables, which sink, what parallelism, what schema-change behaviour. The mechanical glue is inside the connector JARs.
5. The runtime picture also collapses. The legacy stack ran three JVMs (Kafka Connect for Debezium, Kafka broker, Flink task manager). Flink CDC runs one JVM cluster (Flink task manager). Ops on-call goes from three pages to one.

Output.

Metric	Legacy stack	Flink CDC 3.x	Ratio
Lines of config / code	~500	~40	12×
JVM services	3 (Kafka + Kafka Connect + Flink)	1 (Flink)	3×
Distinct checkpointing stories	3 (Debezium offsets, Kafka log, Flink checkpoint)	1 (Flink checkpoint)	3×
Schema-evolution PRs required	4 teams	0 (auto-propagate)	infinite
Median end-to-end latency	300–500 ms	50–100 ms	5×

Rule of thumb. If your ingest pipeline does not fan out to multiple downstream consumers and does not need Kafka's long-retention buffer for replay, the Flink CDC 3.x single-job pattern is the 2026 default. Keep Kafka in the middle only if you have a specific reason (multi-consumer fan-out, cross-team schema contract, or a separate replay window).

Worked example — the operational cost delta between three services and one

Detailed explanation. The infrastructure bill for a mid-size ingest pipeline tells the same story as the line count. A 10-table MySQL → Iceberg pipeline running through Debezium + Kafka + Flink typically runs on ~12 vCPU across three service tiers plus 200 GB of Kafka storage (retention). The equivalent Flink CDC 3.x pipeline runs on ~4 vCPU on a single Flink task manager with zero Kafka storage. The 3× compute delta and the eliminated storage cost are direct — this is not a marketing claim, it's an SRE spreadsheet.

• Legacy 10-table pipeline cost. Kafka broker (3 vCPU), Kafka Connect worker for Debezium (3 vCPU), Flink task manager (6 vCPU), Kafka storage (200 GB × $0.10/GB-month = $20).
• Flink CDC 3.x cost. Flink task manager (4 vCPU), zero Kafka.
• Monthly delta. Roughly 8 vCPU saved plus $20 in Kafka storage. At ~$50/vCPU/month cloud pricing that is $420/month per pipeline. Across 20 pipelines: $8400/month.

Question. A team runs 20 ingest pipelines on the legacy three-hop stack. Quantify the compute + storage cost delta of migrating them all to Flink CDC 3.x and describe how to phase the migration without a big-bang rewrite.

Input.

Component	Legacy vCPU / pipeline	Flink CDC vCPU / pipeline	vCPU delta
Kafka broker share	3	0	-3
Kafka Connect / Debezium worker	3	0	-3
Flink task manager	6	4	-2
Total per pipeline	12	4	-8
20 pipelines	240	80	-160

Code.

# Phase-1 pilot pipeline — new low-risk table pair on Flink CDC 3.x
source:
  type: mysql
  name: pilot-audit-log
  hostname: mysql-audit.internal
  port: 3306
  username: cdc_reader
  password: ${MYSQL_CDC_PASSWORD}
  tables: audit.event

sink:
  type: iceberg
  name: pilot-audit-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: audit.event

pipeline:
  name: pilot-audit-flink-cdc
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

# Cost calculator — one-shot compute of the migration savings
cat <<'EOF' | python3
PIPELINES = 20
LEGACY_VCPU_PER_PIPELINE  = 12
FLINK_CDC_VCPU_PER_PIPELINE = 4
KAFKA_STORAGE_GB_PER_PIPELINE = 200

VCPU_PRICE_MONTH  = 50   # cloud-list-ish
STORAGE_PRICE_GB  = 0.10

legacy_vcpu = PIPELINES * LEGACY_VCPU_PER_PIPELINE
new_vcpu    = PIPELINES * FLINK_CDC_VCPU_PER_PIPELINE
vcpu_saved  = legacy_vcpu - new_vcpu

storage_saved_gb = PIPELINES * KAFKA_STORAGE_GB_PER_PIPELINE

monthly_vcpu_saving    = vcpu_saved * VCPU_PRICE_MONTH
monthly_storage_saving = storage_saved_gb * STORAGE_PRICE_GB
total_monthly = monthly_vcpu_saving + monthly_storage_saving

print(f"vCPU saved:    {vcpu_saved}")
print(f"Storage saved: {storage_saved_gb} GB")
print(f"Monthly $:     {total_monthly}")
print(f"Annual $:      {total_monthly * 12}")
EOF

Step-by-step explanation.

1. The migration does not have to be a big bang. Phase 1: pick a low-risk table pair (an audit log or a slowly-changing lookup table) and stand up a Flink CDC 3.x pipeline in parallel to the legacy Debezium + Kafka pipeline. Both pipelines write to the same sink table (or to a _cdc shadow table) for reconciliation.
2. Phase 2: after two weeks of parallel running with zero divergence, cut over the reads for that table to the Flink CDC pipeline, tear down the Debezium + Kafka pipeline for that source. This releases 8 vCPU immediately.
3. Phase 3: repeat for the next 5 tables. Order matters — start with the least business-critical tables to build ops confidence, then move to the core tables once the runbook is proven.
4. The cost math above shows the annual saving for a 20-pipeline migration at ~$100k in vCPU + storage. The migration itself is roughly 3 engineer-months per pipeline (parallel run, reconciliation, cutover), so the payback period is about 4 months after the first pipeline pair migrates.
5. The hidden win is the ops surface. On-call load drops by ~60% because the entire class of "Kafka rebalance during a Debezium reconfigure" incidents disappears. That saving does not show up in the cost sheet but shows up sharply in on-call satisfaction.

Output.

Phase	Pipelines migrated	vCPU released	Storage released	Monthly $ saved
1 (1 pilot)	1	8	200 GB	~$420
2 (5 tables)	6	48	1.2 TB	~$2400
3 (full migration)	20	160	4 TB	~$8400

Rule of thumb. Migrate one pipeline pair, run in parallel for two weeks with row-level reconciliation, cut over, then repeat. The 20-pipeline migration takes ~6 months of engineer time and pays back within 4 months of the first cutover.

Worked example — when Kafka in the middle still earns its keep

Detailed explanation. The collapse is real for the majority of pipelines, but not for all. Three specific patterns still need Kafka between the CDC source and the analytical sinks: multi-consumer fan-out, cross-team schema contracts, and long-retention replay windows. Recognising the pattern before ripping out Kafka is the senior architectural signal.

• Multi-consumer fan-out. Two or more downstream consumers each want their own view of the change stream (analytics sink + search-index writer + audit archiver). Kafka's log storage decouples the producer from N consumers.
• Cross-team schema contract. Team A owns the source; teams B, C, D own the sinks. Kafka acts as the boundary where the schema contract is defined (Avro schema in Confluent Schema Registry or JSON Schema in Apicurio).
• Replay window. The team needs to be able to re-materialise the sink from N days ago because the sink logic changed. Kafka's log retention gives you a bounded replay window; the source database does not (WAL segments expire quickly).

Question. For a workload that requires fan-out to two consumers (analytics warehouse + search index), design the hybrid architecture that keeps Kafka in the middle but uses Flink CDC to write to Kafka and Flink CDC sinks to consume from Kafka.

Input.

Requirement	Value
Source	MySQL orders table (~5000 events/sec)
Consumer 1	Iceberg data lake (analytics)
Consumer 2	Elasticsearch index (search)
Replay window required	7 days

Code.

# Producer pipeline — MySQL → Kafka via Flink CDC
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: cdc_reader
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.\.*

sink:
  type: kafka
  name: orders-kafka
  properties.bootstrap.servers: kafka.internal:9092
  topic: cdc.orders
  value.format: debezium-json
  properties.compression.type: zstd

pipeline:
  name: mysql-orders-to-kafka
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

# Consumer 1 — Kafka → Iceberg
source:
  type: kafka
  name: orders-from-kafka
  properties.bootstrap.servers: kafka.internal:9092
  topic: cdc.orders
  value.format: debezium-json
  scan.startup.mode: earliest-offset

sink:
  type: iceberg
  name: orders-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: orders.orders

pipeline:
  name: orders-kafka-to-iceberg
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 60s

# Consumer 2 — Kafka → Elasticsearch (search index)
source:
  type: kafka
  name: orders-from-kafka
  properties.bootstrap.servers: kafka.internal:9092
  topic: cdc.orders
  value.format: debezium-json
  scan.startup.mode: earliest-offset

sink:
  type: elasticsearch
  name: orders-search
  hosts: https://es.internal:9200
  index: orders-search
  format: json

pipeline:
  name: orders-kafka-to-elasticsearch
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

Step-by-step explanation.

1. The producer pipeline reads MySQL binlog via Flink CDC and writes Debezium-JSON-formatted events into a Kafka topic cdc.orders. This is the only pipeline that touches MySQL — the source database is protected from N consumer contention.
2. Consumer 1 reads from cdc.orders and writes to Iceberg. Consumer 2 reads from the same topic and writes to Elasticsearch. Each consumer is an independent Flink CDC pipeline with its own YAML file, checkpoint interval, and parallelism.
3. The key difference from the pre-3.x world: Debezium is inside the producer pipeline as the embedded engine — the operator does not see a Kafka Connect worker at all. What the operator sees is one producer YAML and N consumer YAMLs, each an independent Flink job.
4. Kafka retention is set to 7 days (retention.ms = 604800000) which gives both consumers a 7-day replay window. If consumer 2 (Elasticsearch) needs a full re-index after a mapping change, the operator sets scan.startup.mode = earliest-offset in the consumer pipeline YAML and re-runs — Kafka replays the 7 days of events.
5. The op-surface is now: Kafka cluster (1 service, shared across pipelines) + N Flink jobs. This is roughly 1.5× the compute of the pure single-job pattern but with the flexibility to add / remove / replay consumers independently. That trade-off is the correct one for multi-consumer fan-out.

Output.

Pattern	Kafka?	Services	Trade-off
Pure single-job (recommended default)	no	1 Flink cluster	Simplest; no replay window
Producer-broker + N consumers (this pattern)	yes	1 Kafka + N Flink jobs	Multi-consumer + replay; +50% cost
Legacy Debezium + Kafka + Flink (deprecated)	yes	3 tiers, 3 checkpointing stories	Highest cost, highest ops surface

Rule of thumb. Keep Kafka only when you have a concrete second consumer or a hard replay-window requirement. The default choice is the pure single-job pattern; the Kafka-in-the-middle pattern is a deliberate opt-in for fan-out or replay, not a legacy hangover.

Senior interview question on the architectural collapse

A senior interviewer often opens with: "You inherit a legacy ingest stack with Debezium writing to Kafka and a downstream Flink job consuming Kafka to write into Iceberg. Walk me through when — and whether — you'd migrate to Flink CDC 3.x's single-job pipeline, and the specific criteria you'd use to keep Kafka in the middle for particular workloads."

Solution Using a criteria-driven migration decision framework

Decision framework — Flink CDC 3.x single-job vs Kafka-in-the-middle in 2026
============================================================================

Criterion                       Score     Verdict
------------------------------  --------  ---------------------------------
1. Multiple downstream consumers of the change stream?
   yes → keep Kafka                        (multi-consumer fan-out)
   no  → single-job pipeline               (simplest)

2. Replay window required beyond source WAL retention (typically < 12h)?
   yes → keep Kafka                        (retention.ms controls replay)
   no  → single-job pipeline

3. Cross-team schema contract published as a Kafka topic?
   yes → keep Kafka                        (org contract, not technical)
   no  → single-job pipeline

4. Ops team owns Kafka already (mature runbooks + alerting)?
   yes → keep Kafka acceptable             (marginal cost is low)
   no  → single-job pipeline               (avoid new service)

5. Latency SLA sub-200ms end-to-end?
   yes → single-job pipeline               (Kafka adds ~100ms)
   no  → either

Migration path if single-job wins:
  Phase 1: One pilot pipeline (low-risk table) parallel-run for 2 weeks
  Phase 2: Reconciliation query proves zero divergence
  Phase 3: Cut over reads; tear down legacy Debezium + Kafka pipeline
  Phase 4: Repeat for next batch of tables
  Phase 5: Kafka cluster decommission (only if no pipelines still depend on it)

Step-by-step trace.

Question	Answer for typical case	Reasoning
Multi-consumer fan-out?	no	Iceberg is the only consumer
Replay > 12h?	yes	Need 7-day replay for reprocess-on-logic-change
Cross-team schema contract?	no	One team owns source + sink
Kafka mature in the org?	yes	Existing runbooks + alerting
Latency SLA sub-200ms?	no	2-minute freshness SLA is fine
Verdict	keep Kafka in the middle	Replay window + Kafka maturity outweigh single-job simplicity

After the analysis the team decides not to fully collapse — the replay-window requirement pins Kafka in place. But they still migrate the Debezium half to Flink CDC (embedded engine), which removes the Kafka Connect worker JVM. Half a collapse is still a real ops win.

Output:

Layer	Before	After
Debezium / Kafka Connect worker	separate JVM	embedded in Flink CDC source
Kafka broker	separate cluster	retained (replay window)
Downstream Flink consumer	Flink DataStream job	Flink CDC consumer YAML
Total JVM tiers	3	2
Kafka storage	200 GB / pipeline	200 GB / pipeline (unchanged)

Why this works — concept by concept:

• Criteria-driven, not dogma-driven — the migration verdict is a function of the workload's requirements, not "Flink CDC is always better." A senior engineer maps criteria to verdict; a junior engineer proselytises.
• Half-collapses are valid — dropping Debezium's Kafka Connect worker while keeping Kafka is a real intermediate win. The framework does not force all-or-nothing.
• Pilot-then-scale — the migration path starts with one low-risk pipeline in parallel, then scales. Big-bang cutovers of stateful pipelines are how outages happen; parallel-run + reconciliation is how migrations succeed.
• Reconciliation query — cutover is gated on a row-level SELECT ... EXCEPT ... proving the two pipelines produced identical sink state for a 2-week window. The migration is defensible when the reconciliation query returns zero rows.
• Cost — the decision framework is O(1) at design time. The migration cost is O(pipelines) at ~3 engineer-months per pipeline pair, with payback within 4 months of the first cutover for the compute + storage savings alone.

Streaming
Topic — streaming
Streaming ingest and CDC design problems

Practice →

ETL Topic — etl ETL problems on ingest architecture and migration

Practice →

---

2. Source connectors — Postgres / MySQL / Mongo / Oracle

Four source families, four log-tap protocols — knowing each one is the source-connector interview

The mental model in one line: every Flink CDC source connector wraps a protocol-specific log tap (pgoutput logical replication for Postgres, binlog + GTID for MySQL, change streams for MongoDB, LogMiner or XStream for Oracle) plus a two-phase execution model (snapshot then streaming) — and every senior interview conversation about flink cdc postgres or flink cdc mysql is about which knobs you tune per protocol. The four sources look identical from the pipeline YAML — same source: block shape — but the underlying failure modes are wildly different, and mixing them up is the fastest way to fail a senior interview.

The four axes per source.

• Log protocol. Postgres: logical replication via a pgoutput publication and a replication slot. MySQL: binlog rows-format events, anchored by a GTID (Global Transaction ID). MongoDB: change stream cursors driven by resume tokens. Oracle: LogMiner (older, slower) or XStream (newer, licensed).
• Snapshot phase. Every source starts with a consistent snapshot of the initial state, then transitions to streaming from a captured log position. The snapshot mechanism differs — Postgres uses a synchronous slot creation with EXPORT_SNAPSHOT; MySQL uses a global read lock briefly to record the binlog position; Mongo reads the collection under a change-stream watch that begins after the read completes.
• State bookkeeping. Postgres holds a replication slot in the catalog — if the CDC job dies without releasing it, the WAL will not vacuum and the source disk fills. MySQL binlog retention is set at the server level (binlog_expire_logs_seconds); if the CDC job lags past the retention, replay is impossible. Mongo's oplog is capped-collection style with a fixed window; lagging past the window forces a snapshot re-do.
• Failure modes. Each source has a signature failure. Postgres: leaked replication slot. MySQL: GTID gap during a failover. Mongo: oplog rollover mid-stream. Oracle: LogMiner too slow at high change volumes → need XStream.

Postgres source — pgoutput logical replication.

• Protocol. Postgres 10+ supports the pgoutput logical replication plugin natively. Flink CDC creates a publication on the source tables and a replication slot to stream changes.
• Snapshot. On job start, Flink CDC opens a repeatable-read transaction, exports a snapshot ID (EXPORT_SNAPSHOT), reads each table under that snapshot, then transitions to streaming from the corresponding WAL position.
• State. The replication slot lives in pg_replication_slots. It advances with each ack from the CDC job. If the job dies without dropping the slot, the WAL cannot vacuum — the disk fills within hours on a busy source.
• Config knobs. wal_level = logical, max_replication_slots >= 1, max_wal_senders >= 1 on the Postgres side. In the YAML: slot.name, decoding.plugin.name = pgoutput, publication.name.

MySQL source — binlog + GTID.

• Protocol. MySQL 5.7+ emits row-based binlog events. Flink CDC consumes the binlog stream directly using the MySQL replication protocol. GTID mode makes replay across failovers safe — the CDC job resumes from the last GTID rather than a binlog file + position that changes across servers.
• Snapshot. Flink CDC takes a brief global read lock (FLUSH TABLES WITH READ LOCK) or, with scan.incremental.snapshot.enabled = true (the modern default), a lock-free chunked snapshot that splits each table into ranges and reads them in parallel.
• State. The snapshot phase writes chunk-completion markers to a state backend; the streaming phase records the last GTID acked. On restart from savepoint, Flink CDC resumes both.
• Config knobs. server-id (must be unique across all MySQL replication clients — including your CDC job), server-time-zone, scan.incremental.snapshot.enabled = true, scan.startup.mode = initial | latest-offset | specific-offset.

MongoDB source — change streams.

• Protocol. MongoDB 4.0+ exposes change streams — a server-side watch on a collection or database. Flink CDC opens a change-stream cursor and reads change events (insert, update, delete, replace) along with resume tokens.
• Snapshot. Read the collection under a change-stream watch; the snapshot phase reads all documents; the streaming phase begins from the resume token captured at snapshot start.
• State. The resume token identifies a specific position in the oplog. On restart, the CDC job resumes from the last acked resume token.
• Config knobs. hosts (replica set connection string), database, collection, startup.mode = initial | latest-offset | timestamp.

Oracle source — LogMiner or XStream.

• Protocol. LogMiner (built-in, slower, no license required) reads Oracle redo logs and archived redo logs. XStream (Enterprise Edition + Oracle GoldenGate licensed) provides a lower-latency, higher-throughput stream.
• Snapshot. SCN-anchored (System Change Number). On start, Flink CDC records the current SCN, snapshots the tables under AS OF SCN, then streams from that SCN forward.
• State. The SCN cursor. On restart, resume from the last acked SCN.
• Config knobs. url, username, password, database-name, schema-name, log.mining.strategy = online_catalog | redo_log_catalog.

Common interview probes on sources.

• "What happens if the Postgres CDC job crashes and does not drop the replication slot?" — WAL cannot vacuum; disk fills.
• "Why is server-id important for MySQL CDC?" — it must be unique across all replication clients; a collision causes silent event skipping.
• "Can Mongo CDC survive an oplog rollover?" — no; must resnapshot.
• "When do you pick XStream over LogMiner for Oracle CDC?" — high change volume (~10k events/sec+); requires Oracle GoldenGate license.

Worked example — MySQL source with GTID + lock-free snapshot

Detailed explanation. MySQL 8 with GTID enabled is the modern default for MySQL CDC. The lock-free incremental snapshot (introduced in Flink CDC 2.x, hardened in 3.x) reads each table in parallel chunks without holding a global read lock — the source keeps serving traffic during the snapshot. Walk through the YAML config, the server-side GTID requirement, and the failure story if server-id collides with another replica.

• Server-side GTID. gtid_mode = ON, enforce_gtid_consistency = ON, log_bin = ON, binlog_format = ROW, binlog_row_image = FULL.
• CDC-side GTID. scan.startup.mode = initial on first start (snapshot + stream); on savepoint restart, resume from the last acked GTID recorded in Flink state.
• Server-id. Each CDC job needs a unique server-id (or range of IDs for parallel snapshot workers). Colliding IDs cause silent event skipping.

Question. Configure a MySQL source for a table orders.orders with 3 parallel snapshot workers, GTID mode, and a server-id range that will not collide with existing MySQL replicas.

Input.

Parameter	Value
MySQL host	mysql.internal:3306
Database	orders
Table	orders
Existing replica server-ids	1 (master), 2 (replica-1), 3 (replica-2)
CDC parallel snapshot workers	3
Startup mode	initial (snapshot + stream)

Code.

-- Server-side prerequisites — run on the MySQL master as root
-- 1. Confirm GTID + row-format binlog are enabled
SHOW VARIABLES LIKE 'gtid_mode';
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';
-- All three should be ON / ROW / FULL

-- 2. Create the CDC user with the minimum required grants
CREATE USER 'flink_cdc'@'%' IDENTIFIED BY 'strong-secret';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT
  ON *.* TO 'flink_cdc'@'%';
FLUSH PRIVILEGES;

-- 3. Confirm binlog retention is long enough for potential snapshot delay
SHOW VARIABLES LIKE 'binlog_expire_logs_seconds';
-- Recommended: >= 7 days = 604800

# pipeline.yaml — MySQL source with GTID + lock-free snapshot
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  # server-id range — one per parallel snapshot worker
  # avoid 1..3 (already in use); pick 5400..5402
  server-id: 5400-5402
  server-time-zone: UTC
  # Lock-free chunked snapshot
  scan.incremental.snapshot.enabled: true
  scan.incremental.snapshot.chunk.size: 8096
  # Startup mode
  scan.startup.mode: initial
  # Debezium engine tuning
  debezium.snapshot.locking.mode: none

sink:
  type: doris
  name: orders-doris
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}
  table.create.properties.replication_num: 3
  sink.enable-delete: true

pipeline:
  name: orders-mysql-to-doris
  parallelism: 3
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

Step-by-step explanation.

1. The server-side prerequisites are non-negotiable. Without gtid_mode = ON you cannot resume across a failover; without binlog_format = ROW the change events are SQL-statement-based (unusable for CDC); without binlog_row_image = FULL the change event may lack before-image columns (needed for keyed sinks).
2. The CDC user needs REPLICATION SLAVE and REPLICATION CLIENT to open the binlog stream, plus SELECT and RELOAD for the snapshot phase. Nothing more — do not grant SUPER or ALL PRIVILEGES.
3. The server-id: 5400-5402 range gives three unique IDs, one per parallel snapshot worker (parallelism: 3). The IDs must not collide with the master's or any existing replica's IDs — otherwise both the CDC job and the colliding replica get inconsistent binlog streams.
4. scan.incremental.snapshot.enabled: true activates the lock-free chunked snapshot. Each table is split into ranges by primary key; each parallel worker reads one chunk at a time; no global read lock is taken. The source database serves user traffic normally during the snapshot.
5. scan.startup.mode: initial says "snapshot then stream" on first job start. On savepoint restart, this setting is ignored — the job resumes from the state recorded in the checkpoint. This is a common gotcha: developers set scan.startup.mode = latest-offset after a restart and lose the snapshot state.

Output.

Phase	Duration (10M-row table)	Source impact
Snapshot (3 parallel workers, lock-free)	~5 minutes	~10% CPU, no locks
Streaming (steady state, 500 ev/s)	ongoing	negligible binlog decode CPU
Restart from savepoint	~15 seconds	resume from last GTID

Rule of thumb. For MySQL CDC in 2026, always turn on GTID + row-format binlog on the source, always use the lock-free incremental snapshot, and always allocate a server-id range that gives one unique ID per snapshot worker plus a couple of spares.

Worked example — Postgres source with pgoutput and slot hygiene

Detailed explanation. Postgres CDC via pgoutput is elegant when configured correctly and a full-cluster outage when misconfigured. The replication slot lives in the Postgres catalog; if the CDC job dies without dropping the slot, the WAL cannot vacuum, disk fills, autovacuum stops, and the source database gradually degrades to unusable. Walk through the correct config, the drop-slot hygiene, and the alert query that catches a stale slot before disk pressure.

• Server-side. wal_level = logical, max_replication_slots >= 5, max_wal_senders >= 5.
• Publication + slot. Flink CDC creates a publication (list of tables to replicate) and a slot (position in the WAL). Both live in the catalog.
• Hygiene. Every "goodbye" of a CDC job must drop the slot, else the WAL retention grows without bound.
• Alert. SELECT slot_name, active, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes FROM pg_replication_slots — alert when lag_bytes > 10 GB.

Question. Configure a Postgres source for the public.orders table with slot hygiene, and add the monitoring query + alert that catches a stale slot before it fills disk.

Input.

Parameter	Value
Postgres host	pg-primary.internal:5432
Database	ordersdb
Table	public.orders
Slot name	flink_cdc_orders
Publication name	flink_cdc_pub
Alert threshold	slot lag > 10 GB

Code.

-- Server-side prerequisites — postgresql.conf
-- wal_level              = logical
-- max_replication_slots  = 5
-- max_wal_senders        = 5
-- max_worker_processes   = 8

-- Create the CDC role with the minimum required privileges
CREATE ROLE flink_cdc WITH LOGIN REPLICATION PASSWORD 'strong-secret';
GRANT CONNECT ON DATABASE ordersdb TO flink_cdc;
GRANT USAGE  ON SCHEMA  public     TO flink_cdc;
GRANT SELECT ON public.orders      TO flink_cdc;

-- Publication (create manually so ownership is explicit)
CREATE PUBLICATION flink_cdc_pub FOR TABLE public.orders;

-- Verify slot list (empty at this point — Flink CDC will create the slot)
SELECT slot_name, plugin, active FROM pg_replication_slots;

-- Alert query — run every 30 seconds via Prometheus scraper
SELECT slot_name,
       active,
       pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes,
       age(now(), stats_reset)                                    AS slot_age
FROM   pg_replication_slots
WHERE  plugin = 'pgoutput';
-- Alert when: lag_bytes > 10737418240   (10 GB)
--         or: active = false AND slot_age > interval '5 minutes'

# pipeline.yaml — Postgres CDC with slot hygiene
source:
  type: postgres
  name: orders-postgres
  hostname: pg-primary.internal
  port: 5432
  username: flink_cdc
  password: ${PG_CDC_PASSWORD}
  database-name: ordersdb
  schema-name: public
  tables: orders
  slot.name: flink_cdc_orders
  decoding.plugin.name: pgoutput
  publication.name: flink_cdc_pub
  # Snapshot options
  scan.incremental.snapshot.enabled: true
  scan.startup.mode: initial
  # Debezium engine tuning
  debezium.publication.autocreate.mode: filtered
  debezium.snapshot.mode: initial

sink:
  type: iceberg
  name: orders-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: orders.orders

pipeline:
  name: orders-postgres-to-iceberg
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

# Manual slot cleanup — run only after confirming the CDC job is stopped
psql -h pg-primary.internal -U dba -d ordersdb <<'EOF'
-- Confirm the slot is not active before dropping
SELECT slot_name, active FROM pg_replication_slots WHERE slot_name = 'flink_cdc_orders';
-- If active = false:
SELECT pg_drop_replication_slot('flink_cdc_orders');
-- If active = true, the CDC job is still running — do NOT drop
EOF

Step-by-step explanation.

1. The server-side prerequisites (wal_level = logical, max_replication_slots, max_wal_senders) are one-time and reboot-required. Get them right on day one; the alternative is a full-cluster restart to enable them later.
2. The CDC role has only REPLICATION, CONNECT, USAGE, and SELECT on the CDC tables. No SUPERUSER. If the CDC user is compromised, blast radius is limited to reading the replicated tables.
3. debezium.publication.autocreate.mode: filtered says: create a publication that includes only the tables specified in the YAML tables: list. The alternative (all_tables) publishes every table on the database — a security and performance footgun.
4. The alert query catches two failure modes. First: lag_bytes > 10 GB means the CDC job is behind on ack — probably because the sink is slow or the CDC job is throttled. Second: active = false AND slot_age > 5m means the slot is leaked (CDC job died without dropping it) — the WAL is accumulating and disk pressure is imminent.
5. The manual cleanup at the end is the ops runbook. When retiring a pipeline, always: (a) stop the Flink job, (b) verify the slot is active = false, (c) drop the slot. Skipping step (c) is the #1 Postgres CDC production incident.

Output.

Slot health	lag_bytes	active	Alert	Action
Healthy	< 100 MB	true	none	none
Lagging	5 GB	true	warn	investigate sink slowness
Saturated	12 GB	true	page	scale sink; consider throttling source
Leaked	20 GB	false	page (P0)	verify job stopped; drop slot

Rule of thumb. Every Postgres CDC pipeline ships with the slot-lag alert on day one. The slot is a database-catalog resource; forgetting to drop it after a pipeline retirement is the CDC equivalent of leaving an SSH key on a public repo.

Worked example — MongoDB source with resume tokens

Detailed explanation. MongoDB CDC via change streams is the simplest of the four protocols on the happy path and the most opaque on the failure path. Change streams give the CDC job a stream of insert / update / delete / replace events with resume tokens; the CDC job stores the last acked resume token in Flink state. The subtle failure mode: if the CDC job lags past the oplog window (typically 24 hours on a busy replica set), the resume token is invalidated and the only recovery is a full re-snapshot. Walk through the correct config plus the lag-monitor query.

• Oplog basics. MongoDB's oplog is a capped collection (fixed size). Old events roll off when new events push past the size cap. The retention window is the oplog size divided by the write rate.
• Resume token. An opaque token that identifies a position in the oplog. Valid until the corresponding event rolls off the oplog.
• Failure mode. Resume token invalid → CDC job cannot resume → full snapshot required.

Question. Configure a MongoDB source for the orders.orders collection with resume-token persistence, and add the oplog-lag alert that catches a stale CDC job before the oplog rolls over.

Input.

Parameter	Value
Mongo hosts	mongo-rs.internal:27017
Database	orders
Collection	orders
Replica set name	rs0
Oplog size	100 GB
Alert threshold	lag > 6 hours (oplog window is ~24h)

Code.

// Server-side prerequisites — confirm oplog size and write rate
// Run from mongosh
use local
db.oplog.rs.stats().maxSize          // capped collection size in bytes
db.oplog.rs.stats().count            // number of oplog entries
db.oplog.rs.stats().size             // current used size

// Compute retention window (seconds)
var oldest = db.oplog.rs.find().sort({ts:1}).limit(1).next().ts.getTime()/1000
var newest = db.oplog.rs.find().sort({ts:-1}).limit(1).next().ts.getTime()/1000
print("Oplog window seconds:", newest - oldest)

// Create the CDC role with the minimum required privileges
use admin
db.createUser({
  user: "flink_cdc",
  pwd:  "strong-secret",
  roles: [
    { role: "read", db: "orders" },
    { role: "read", db: "local"  }   // required for oplog / change streams
  ]
})

# pipeline.yaml — MongoDB CDC with change stream
source:
  type: mongodb
  name: orders-mongo
  hosts: mongo-rs.internal:27017
  username: flink_cdc
  password: ${MONGO_CDC_PASSWORD}
  # Connection options
  connection.options: replicaSet=rs0&readPreference=secondaryPreferred
  database: orders
  collection: orders
  # Snapshot options
  scan.startup.mode: initial
  copy.existing: true
  copy.existing.max.threads: 4
  copy.existing.queue.size: 16000
  # Change stream tuning
  poll.max.batch.size: 1000
  poll.await.time.ms: 1500
  heartbeat.interval.ms: 30000

sink:
  type: kafka
  name: orders-kafka
  properties.bootstrap.servers: kafka.internal:9092
  topic: cdc.orders
  value.format: debezium-json

pipeline:
  name: orders-mongo-to-kafka
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

// Lag-monitor query — run every 60s from Prometheus scraper
// The Flink CDC job publishes the last acked resume token timestamp
// as a metric; Mongo oplog head timestamp is queried here.
use local
var latestOplog = db.oplog.rs.find().sort({ts:-1}).limit(1).next().ts.getTime()/1000
print("latest_oplog_epoch_s:", latestOplog)

// Emit as a Prometheus gauge; alert:
//   time.time() - flink_cdc_mongo_last_ack_ts > 21600   (6 hours)

Step-by-step explanation.

1. The oplog window computation is critical. A 100 GB oplog on a source writing 1 MB/s of oplog data has a ~100000-second (~28-hour) window. The CDC job must ack change events well within that window; the alert threshold is roughly 25% of the window.
2. The CDC user needs read on the source database and on local (which contains the oplog). No admin privileges. Change streams are exposed via a normal read path.
3. copy.existing: true runs the snapshot phase before streaming. copy.existing.max.threads: 4 reads the collection in parallel. The change stream watch starts at the snapshot's start time, so no events are missed between snapshot and streaming.
4. heartbeat.interval.ms: 30000 makes the change stream emit an idle heartbeat every 30 seconds even when no changes occur. Without heartbeats, a completely-idle collection would let the CDC job's resume token drift toward oplog expiry.
5. The lag monitor compares the CDC job's last-acked resume-token timestamp against the current oplog head. If the delta exceeds 6 hours (25% of the 24-hour window), page — the CDC job is at real risk of losing its position.

Output.

Metric	Healthy	Warn	Page
Oplog window	~24h	—	—
CDC lag (last ack)	< 5 min	6h–12h	> 12h
Snapshot progress	100% at start	—	stuck
Change stream heartbeats	every 30s	dropped for 5m	dropped for 15m

Rule of thumb. For MongoDB CDC, always know your oplog window, always ship heartbeats, and always alert at 25% of the window on lag. Losing the resume token is a full re-snapshot — an expensive recovery on a large collection.

Senior interview question on source-connector selection

A senior interviewer might ask: "You inherit an ingest requirement — three sources (Postgres, MySQL, Mongo) all landing in Iceberg. Walk me through the source-connector configuration for each, the specific protocol knobs, the failure mode you monitor per source, and the shared vs per-source infrastructure decisions."

Solution Using a per-source config + shared Flink cluster architecture

# pipeline-orders-postgres.yaml — Postgres source
source:
  type: postgres
  name: orders-postgres
  hostname: pg-primary.internal
  port: 5432
  username: flink_cdc
  password: ${PG_CDC_PASSWORD}
  database-name: ordersdb
  schema-name: public
  tables: orders
  slot.name: flink_cdc_orders
  decoding.plugin.name: pgoutput
  publication.name: flink_cdc_pub
  scan.incremental.snapshot.enabled: true
  scan.startup.mode: initial

sink:
  type: iceberg
  name: orders-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: orders.orders

pipeline:
  name: orders-postgres-to-iceberg
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

# pipeline-users-mysql.yaml — MySQL source
source:
  type: mysql
  name: users-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: users.\.*
  server-id: 5500-5502
  server-time-zone: UTC
  scan.incremental.snapshot.enabled: true
  scan.startup.mode: initial

sink:
  type: iceberg
  name: users-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: users.\.*

pipeline:
  name: users-mysql-to-iceberg
  parallelism: 3
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

# pipeline-events-mongo.yaml — MongoDB source
source:
  type: mongodb
  name: events-mongo
  hosts: mongo-rs.internal:27017
  username: flink_cdc
  password: ${MONGO_CDC_PASSWORD}
  connection.options: replicaSet=rs0&readPreference=secondaryPreferred
  database: events
  collection: events
  scan.startup.mode: initial
  copy.existing: true
  copy.existing.max.threads: 4
  heartbeat.interval.ms: 30000

sink:
  type: iceberg
  name: events-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: events.events

pipeline:
  name: events-mongo-to-iceberg
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

Step-by-step trace.

Source	Protocol	Snapshot	Failure to monitor	Config knob signal
Postgres	pgoutput logical replication	EXPORT_SNAPSHOT + WAL	slot lag / leaked slot	slot.name, publication.name
MySQL	binlog + GTID	lock-free chunked	binlog expiry / server-id collision	server-id range, scan.incremental.snapshot.enabled
Mongo	change stream + resume tokens	copy.existing + change-stream watch	oplog rollover past resume token	heartbeat.interval.ms, copy.existing.max.threads

After the rollout, the three pipelines run as three independent Flink jobs on the shared Flink cluster (one cluster runs many jobs; per-source resource isolation is achieved via slot sharing groups). Each job has its own YAML file, its own checkpoint state, and its own on-call metric. The shared Flink cluster is the platform; the per-source config is the pipeline.

Output:

Aspect	Postgres pipeline	MySQL pipeline	Mongo pipeline
Log tap	pgoutput slot	binlog + GTID	change stream
Snapshot RPO	~5 min for 10M rows	~5 min lock-free chunked	~10 min for 5M docs
Steady-state event rate	500 ev/s	2000 ev/s	1500 ev/s
Alert	slot lag > 10 GB	binlog expiry < 24h	CDC lag > 6h
Rebuild cost if lost position	rebuild from Iceberg (7-day retention)	rebuild from binlog if in retention	full re-snapshot

Why this works — concept by concept:

• Per-source config, shared cluster — every source protocol has its own knobs. The shared Flink cluster hosts N jobs; the YAML per source names the protocol-specific behaviours (slot, server-id, resume token). Do not conflate cluster and job.
• Same sink family (Iceberg) — using one sink across the three sources means one governance / retention / partitioning story for the target. Complexity moves to the source YAML; the sink is uniform.
• Per-source failure monitor — Postgres slot lag, MySQL binlog expiry, Mongo oplog window. Each has a distinct metric with a distinct alert threshold; the shared Grafana dashboard has three panels, not one.
• Minimal privilege on the source — REPLICATION for MySQL, REPLICATION role for Postgres, read on local for Mongo. Never grant admin; never grant more tables than the pipeline replicates.
• Cost — each Flink job costs 2–3 task slots (~1 vCPU + 2 GB heap per slot). Three pipelines on one shared cluster = 6–9 slots. Adding a fourth source is a new YAML, not a new cluster.

Streaming
Topic — streaming
Streaming source-connector interview problems

Practice →

SQL Topic — sql SQL problems on WAL / binlog log protocols

Practice →

---

3. Pipeline YAML — declarative topology

Four YAML sections — source, transform, route, sink — replace 500 lines of DataStream Java

The mental model in one line: the Flink CDC 3.x pipeline YAML has exactly four functional sections (source, transform, route, sink) plus a pipeline header, and understanding the semantics of each section — especially the difference between transform (per-row projection/filter) and route (many-tables-to-one-topic or one-topic-to-many-tables mapping) — is the DSL half of the interview. The YAML is a facade over the Flink DataStream API; every YAML pipeline compiles down to a DataStream graph you could write by hand. But the compression is real — a 40-line YAML replaces ~500 lines of Java for the majority of ingest use cases.

The four functional sections.

• source. Names the source connector (type: mysql | postgres | mongodb | oracle), credentials, table list. Emits Event records that carry both the row data and the schema changes.
• transform. Per-row projection, filter, and derived column expressions. Uses a small SQL-like syntax with a bounded expression grammar (not full ANSI SQL). Runs on the Flink task manager as a stateless map.
• route. Table-to-table remapping. Common uses: rename tables (orders.orders → orders_orders), merge multiple source tables into one sink table (orders.orders, orders.orders_archive → orders.orders_unified), or split one source into many sinks (rare).
• sink. Names the sink connector (type: doris | iceberg | paimon | snowflake | kafka), credentials, target catalog / database / table. Handles the transactional commit and the schema-evolution DDL.
• pipeline. Header — name, parallelism, checkpoint interval, schema-change behaviour policy, and Flink runtime overrides.

The transform section — projection, filter, derived columns.

• Projection. projection: id, name, upper(email) as email_upper — a bounded SQL-like expression list; each entry becomes a column in the downstream event.
• Filter. filter: status <> 'DELETED' — a boolean expression; rows failing the filter are dropped from the stream (not sent to the sink).
• Derived columns. projection: id, name, coalesce(created_at, updated_at) as event_time — computed columns using a small set of built-in functions.
• What is not in the YAML. UDFs, joins, windowed aggregations. The moment you need those, fall back to Flink SQL or Flink DataStream; the YAML DSL intentionally does not go there.

The route section — table remapping.

• Simple rename. route: [{source-table: orders.orders, sink-table: orders_orders}] — write the source-side orders.orders into the sink-side orders_orders. Useful when the sink has a different naming convention.
• Many-to-one merge. route: [{source-table: orders.\.*, sink-table: orders_unified}] — every table matching the regex in the source lands in a single sink table. Requires all source tables to have the same schema (Flink CDC will error at planning time if not).
• One-to-many split. Rare; requires a sink-table expression that includes a routing key. Typically better handled by a downstream Flink SQL job.

The pipeline header — non-negotiable defaults.

• name. Human-readable job name; shows up in the Flink UI.
• parallelism. Number of task slots per operator. Match to source-side snapshot workers plus a small headroom.
• schema.change.behavior. lenient (default) auto-applies compatible DDL to the sink; strict fails the job on any DDL; other modes control ADD/DROP/RENAME independently.
• execution.checkpointing.interval. Frequency of Flink checkpoints. 30 seconds is the default; increase to 60 s if the sink is a slow-committer (Iceberg, Snowflake); decrease to 10 s if the SLA is sub-minute freshness and the sink is fast (Doris, StarRocks).

Common interview probes on the YAML DSL.

• "When does the YAML DSL run out of expressive power?" — UDFs, joins, windowed aggregates.
• "How does route differ from transform?" — route is table-level; transform is row-level.
• "What's the difference between parallelism in the source and the pipeline header?" — pipeline header is the default for all operators; source can override.
• "Can I set per-sink schema.change.behavior?" — no; it's a pipeline-level setting in 3.x.

Worked example — 40-line MySQL → Doris YAML with transform + route

Detailed explanation. The reference pipeline for the "MySQL to Doris" tutorial takes exactly 40 lines — a source, a transform that filters out soft-deleted rows and derives an event-time column, a route that merges two source tables into one sink table, and a Doris sink. Walk through the four sections and explain each choice.

• Source. MySQL orders.orders + orders.orders_archive regex-matched.
• Transform. Filter status <> 'DELETED', project standard columns, derive event_time from created_at | updated_at.
• Route. Merge both source tables into orders_unified in the sink.
• Sink. Doris orders_unified table, with StreamLoad transactional writes.

Question. Produce the full 40-line pipeline.yaml for this workload with explicit annotations on why each line is needed.

Input.

Section	Requirement
Source	MySQL, tables matching orders.orders_.*
Transform	filter deleted; derive event_time
Route	merge into orders_unified
Sink	Doris, transactional stream_load
SLA	60-second freshness

Code.

# pipeline.yaml — 40-line MySQL → Doris with transform + route
source:
  type: mysql
  name: orders-mysql-src
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders_.*             # matches orders.orders, orders.orders_archive
  server-id: 5400-5403
  server-time-zone: UTC
  scan.incremental.snapshot.enabled: true
  scan.startup.mode: initial

transform:
  - source-table: orders.orders_.*
    projection: |
      id,
      customer_id,
      product_id,
      quantity,
      price,
      status,
      created_at,
      updated_at,
      coalesce(updated_at, created_at) as event_time
    filter: status <> 'DELETED'

route:
  - source-table: orders.orders_.*
    sink-table: analytics.orders_unified
    description: >
      Merge live + archived orders into a single sink table for the
      unified analytics view.

sink:
  type: doris
  name: orders-doris-sink
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}
  table.create.properties.replication_num: 3
  sink.enable-delete: true
  sink.buffer-size: 10000
  sink.buffer-count: 3

pipeline:
  name: mysql-orders-to-doris-unified
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

Step-by-step explanation.

1. The source block names the MySQL instance and the regex orders.orders_.* matching both orders.orders and orders.orders_archive. server-id: 5400-5403 gives four unique IDs, matching the pipeline parallelism of 4.
2. The transform block is a list keyed on source-table (a regex) — each list entry applies to matching tables. The projection selects the columns to forward and derives event_time = coalesce(updated_at, created_at). The filter drops soft-deleted rows before they reach the sink.
3. The route block remaps the source tables to a single sink table analytics.orders_unified. Because both source tables share the same schema (post-transform), the merge is safe — Flink CDC verifies this at pipeline start and errors immediately if schemas diverge.
4. The sink block names the Doris FE endpoint, credentials, and StreamLoad-specific properties. sink.enable-delete: true propagates DELETE events (essential for consistency); sink.buffer-size and sink.buffer-count control the micro-batch cadence into Doris.
5. The pipeline header ties it together: name, parallelism, schema.change.behavior: lenient (auto-apply DDL to the sink), and execution.checkpointing.interval: 30s (checkpoint every 30 s to meet the 60-second freshness SLA).

Output.

Section	Lines	Purpose
source	10	MySQL connection + snapshot config
transform	12	Filter + projection + derived column
route	5	Merge two source tables into one sink table
sink	8	Doris FE + StreamLoad tuning
pipeline	4	Name, parallelism, schema policy, checkpoint
Total	~40	Full ingest pipeline

Rule of thumb. The 40-line YAML is the canonical benchmark. If your ingest pipeline needs more than 40 lines of YAML, you probably need Flink SQL (for joins / windows / UDFs) or Flink DataStream (for arbitrary Java). The YAML DSL is intentionally bounded.

Worked example — transform expression syntax

Detailed explanation. The transform section is where beginners most often trip. The expression grammar is small — a subset of ANSI SQL — with roughly 30 built-in functions. It is not Flink SQL. Understanding the grammar boundary saves hours of "why doesn't my UDF work in the YAML."

• Supported. Column references, literals, arithmetic, boolean logic, coalesce, case when, cast, upper / lower, substr, concat, regexp_replace, date_format, to_timestamp, md5, if.
• Not supported. UDFs, joins, window functions, aggregations, subqueries, table-valued functions.
• Escape hatch. If you need something the grammar does not support, precede the CDC pipeline with a source-side view (Postgres / MySQL side) or follow it with a Flink SQL job.

Question. Extract the credit-card BIN (first 6 digits) as a new column, mask the rest, filter out test transactions, and lowercase the currency code — using only the YAML transform expression syntax.

Input.

Source column	Type	Sample
id	BIGINT	42
card_pan	VARCHAR(19)	4111111111111111
amount	DECIMAL(10,2)	99.99
currency	CHAR(3)	USD
is_test	BOOLEAN	false

Code.

transform:
  - source-table: payments.transactions
    projection: |
      id,
      substr(card_pan, 1, 6) as card_bin,
      concat(substr(card_pan, 1, 6),
             '******',
             substr(card_pan, length(card_pan) - 3, 4)) as card_masked,
      amount,
      lower(currency) as currency_code,
      case
        when amount > 10000 then 'HIGH'
        when amount > 1000  then 'MED'
        else                     'LOW'
      end as amount_bucket,
      md5(concat(cast(id as string), '|', card_pan)) as event_id
    filter: is_test = false and amount > 0

Step-by-step explanation.

1. substr(card_pan, 1, 6) as card_bin extracts the first 6 characters of the PAN — the standard "Bank Identification Number" that is safe to store. substr is one of the supported string functions.
2. The masked PAN is built by concatenating the first 6 digits, a fixed ****** mask, and the last 4 digits. length() gives the total length so the slice adapts to different card lengths.
3. case when ... then ... else ... end bucketises the amount. This is the closest the YAML grammar gets to conditional logic; nested case is supported.
4. md5(concat(cast(id as string), '|', card_pan)) produces a deterministic event ID for idempotent sink writes. cast and concat cover the type-coercion boundary; md5 is one of the ~30 built-ins.
5. The filter drops test transactions and any zero-value row. Boolean expressions can combine and, or, not; comparison operators are the usual = != < > <= >=.

Output.

Input row	Output columns
id=42, card_pan=4111111111111111, amount=250.00, currency=USD, is_test=false	id=42, card_bin=411111, card_masked=411111******1111, amount=250.00, currency_code=usd, amount_bucket=LOW, event_id=
id=43, card_pan=..., is_test=true	filtered out

Rule of thumb. Push the projection into YAML transform whenever the expression stays inside the grammar. The moment you need a UDF, a join, or a window, add a downstream Flink SQL job — do not try to smuggle logic into the YAML grammar.

Worked example — route for shard consolidation

Detailed explanation. A classic use of route is consolidating N sharded source tables into 1 sink table. If the source Postgres cluster has orders_shard_0, orders_shard_1, orders_shard_2, orders_shard_3 (all with identical schema), a single route entry maps them all into a unified orders sink table. The YAML makes this a two-line change.

• Source shards. 4 tables with identical schema, sharded on customer_id % 4.
• Sink target. 1 unified orders table.
• Trade-off. The unified sink loses shard identity unless a shard_id column is derived.

Question. Produce the pipeline YAML that consolidates 4 sharded Postgres source tables into a single Iceberg sink, preserving the shard number as a derived column.

Input.

Source table	Rows / hour
orders_shard_0	5000
orders_shard_1	4800
orders_shard_2	5100
orders_shard_3	4900

Code.

source:
  type: postgres
  name: orders-postgres
  hostname: pg-primary.internal
  port: 5432
  username: flink_cdc
  password: ${PG_CDC_PASSWORD}
  database-name: ordersdb
  schema-name: public
  tables: orders_shard_[0-3]
  slot.name: flink_cdc_orders_shards
  decoding.plugin.name: pgoutput
  publication.name: flink_cdc_orders_pub
  scan.incremental.snapshot.enabled: true

transform:
  - source-table: public.orders_shard_0
    projection: "id, customer_id, amount, created_at, 0 as shard_id"
  - source-table: public.orders_shard_1
    projection: "id, customer_id, amount, created_at, 1 as shard_id"
  - source-table: public.orders_shard_2
    projection: "id, customer_id, amount, created_at, 2 as shard_id"
  - source-table: public.orders_shard_3
    projection: "id, customer_id, amount, created_at, 3 as shard_id"

route:
  - source-table: public.orders_shard_[0-3]
    sink-table: analytics.orders_unified
    description: Consolidate 4 shards into a single sink table

sink:
  type: iceberg
  name: orders-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: analytics.orders_unified

pipeline:
  name: postgres-shards-to-iceberg
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 60s

Step-by-step explanation.

1. Four separate transform entries — one per shard — each derives a constant shard_id column matching the shard index. The transform runs per source table, so the constant is per-table.
2. The route block maps all four shards to the single sink table analytics.orders_unified. Because the transform gave each shard a shard_id column, downstream queries can partition or filter by shard.
3. parallelism: 4 matches the number of source shards; Flink CDC allocates one task slot per shard for the snapshot phase, which parallelises the initial backfill.
4. Iceberg is the natural sink for this pattern because Iceberg supports partitioning by shard_id — the query engine can prune to a single partition when a query filters WHERE shard_id = 2.
5. On schema evolution: if one of the shards adds a column, schema.change.behavior: lenient propagates the ADD to the Iceberg sink. If the shards evolve at different times, the sink schema is the union (nullable columns for shards that have not yet added them).

Output.

Source table	Sink table	Rows / hour	Partition on sink
orders_shard_0	orders_unified	5000	shard_id=0
orders_shard_1	orders_unified	4800	shard_id=1
orders_shard_2	orders_unified	5100	shard_id=2
orders_shard_3	orders_unified	4900	shard_id=3
Total	1 table	19800	4 partitions

Rule of thumb. Use route for many-to-one consolidation when the sources have the same schema. Use transform to derive a shard_id or source_table_name column so downstream queries can partition or filter by origin. This pattern replaces the manual "union all + shard column" logic you would otherwise write in downstream Flink SQL.

Senior interview question on the YAML DSL

A senior interviewer might ask: "Design the pipeline YAML for a workload that reads MySQL users + orders tables, filters test users, derives an event_time column, and writes to two different sinks — Iceberg (for analytics) and Kafka (for the search team). Walk me through the YAML sections you'd use, where the YAML runs out of expressive power, and how you'd hand off to Flink SQL if needed."

Solution Using two Flink CDC pipelines sharing one source

# pipeline-users-iceberg.yaml — MySQL users + orders → Iceberg
source:
  type: mysql
  name: users-orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: (users.users|orders.orders)
  server-id: 5600-5603
  server-time-zone: UTC
  scan.incremental.snapshot.enabled: true

transform:
  - source-table: users.users
    projection: |
      id, email, name, created_at, updated_at,
      coalesce(updated_at, created_at) as event_time
    filter: is_test = false
  - source-table: orders.orders
    projection: |
      id, user_id, amount, status, created_at, updated_at,
      coalesce(updated_at, created_at) as event_time
    filter: status <> 'DELETED'

sink:
  type: iceberg
  name: users-orders-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: analytics.\.*

pipeline:
  name: users-orders-mysql-to-iceberg
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

# pipeline-users-kafka.yaml — MySQL users → Kafka (for search team)
source:
  type: mysql
  name: users-mysql-for-search
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: users.users
  server-id: 5604
  server-time-zone: UTC
  scan.incremental.snapshot.enabled: true

transform:
  - source-table: users.users
    projection: |
      id, email, name, created_at, updated_at,
      md5(email) as email_hash
    filter: is_test = false

sink:
  type: kafka
  name: users-kafka
  properties.bootstrap.servers: kafka.internal:9092
  topic: cdc.users
  value.format: debezium-json

pipeline:
  name: users-mysql-to-kafka-search
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s

Step-by-step trace.

Aspect	Iceberg pipeline	Kafka pipeline
Source tables	users.users, orders.orders	users.users only
Transform	filter + derive event_time	filter + hash email
Sink	analytics..* (Iceberg)	cdc.users (Kafka)
Server-id	5600-5603 (4 slots)	5604 (1 slot)
Parallelism	4	2

After deploy, both pipelines run as independent Flink jobs on the shared Flink cluster. Each has its own MySQL binlog stream (each has its own server-id), its own transform logic, and its own sink. When the users table adds a column, schema.change.behavior: lenient propagates the ADD to both sinks independently. When the search team needs a re-index, the Kafka pipeline can be re-run from scan.startup.mode: initial without disturbing the Iceberg pipeline.

Output:

Aspect	Result
Distinct MySQL binlog readers	2 (one per pipeline)
MySQL server-id collision risk	none — separate IDs
Schema evolution propagation	both sinks, independent
Independent replay	yes (per pipeline)
Line count	~40 lines per pipeline

Why this works — concept by concept:

• Two YAMLs, one source — each pipeline opens its own binlog stream with its own server-id. This is the correct pattern; sharing a single binlog reader across pipelines is not supported and would create failure-mode coupling.
• Transform per pipeline — the Iceberg sink wants the raw columns; the Kafka (search) sink wants an email hash. Different transforms per sink is the natural separation.
• Schema-change propagation to both sinks — because both pipelines have schema.change.behavior: lenient, an ADD COLUMN on the source propagates to Iceberg and to the Kafka topic schema (Debezium-JSON auto-includes the new column). No coordination between sinks.
• Where the YAML runs out — if the search team wanted joins (users + orders in one Kafka payload), the YAML cannot express it. Escape hatch: a Flink SQL job downstream of the Kafka topic joins the two streams and writes to a second Kafka topic. Do not try to force joins into YAML.
• Cost — each Flink CDC job costs 2–4 task slots. Two pipelines = 6 slots. The MySQL side sees two independent binlog reader connections, negligible incremental cost.

ETL
Topic — etl
ETL pipeline DSL and topology design problems

Practice →

Design Topic — design Design problems on pipeline configuration and routing

Practice →

---

4. Schema evolution + at-least-once vs EOS

Auto-propagate DDL from source to sink — the killer Flink CDC 3.x feature, with a behaviour matrix per column op

The mental model in one line: Flink CDC 3.x's schema-evolution feature reads DDL events from the source log, buffers them behind the last data event, and issues the equivalent DDL on the sink before the next data event that depends on the new schema — and the schema.change.behavior pipeline setting picks a behaviour matrix (lenient auto-applies compatible ops, strict fails the job on any DDL) that decides how strict the propagation is per column operation. Combined with Flink's checkpointing model plus per-sink transactional writes, this is what gives Flink CDC end-to-end exactly-once semantics — the property that makes it credible for production analytics.

How schema evolution flows.

• Source detects DDL. Postgres logical decoding emits RELATION messages; MySQL binlog carries QUERY events for DDL; Mongo change streams do not have schema DDL (schema is per-document, handled at the sink).
• DDL enters the Flink graph. The source connector emits a special SchemaChangeEvent alongside the normal DataChangeEvent stream. The two streams travel together, ordered by log position.
• Coordinator handles DDL. The pipeline coordinator receives the SchemaChangeEvent, applies the schema.change.behavior policy, and — for lenient mode — issues the equivalent DDL to the sink via the sink connector's MetadataApplier.
• Sink accepts DDL. Doris, StarRocks, Paimon, Iceberg all support online ALTER TABLE ADD COLUMN. The sink connector calls the target's DDL API, waits for success, then resumes data writes.

The schema.change.behavior matrix.

• lenient (default). Auto-apply compatible DDL: ADD COLUMN yes, DROP COLUMN treated as a "column becomes NULL" (data preserved), RENAME COLUMN treated as ADD + DROP (both columns coexist), ALTER COLUMN TYPE on compatible widening (INT → BIGINT) yes, incompatible narrowing (BIGINT → INT) fails.
• strict. No DDL is applied. Any DDL event fails the pipeline. Use when the sink schema is externally managed (dbt, Alembic).
• ignore. DDL events are silently swallowed. Downstream sinks continue to receive data with the old schema; new columns are dropped. Use only when the sink cannot support DDL.
• evolve. Some variants of Flink CDC expose an evolve mode that auto-applies more aggressive DDL. Check the version.

Per column op behaviour under lenient.

• ADD COLUMN. Auto-applied. Sink runs the equivalent ALTER TABLE ADD COLUMN; new rows carry the new column; old rows have NULL for the new column.
• DROP COLUMN. Not physically applied on the sink; the column is instead marked as "no longer written by CDC." Existing sink data is preserved. This is the safe default — dropping columns loses history irreversibly.
• RENAME COLUMN. Applied as ADD new_name + stop writing to old_name. Both columns coexist on the sink after the rename; old_name has NULL for new rows.
• ALTER COLUMN TYPE. Widening (INT → BIGINT, VARCHAR(50) → VARCHAR(100)) auto-applied. Narrowing fails.
• ADD PRIMARY KEY / DROP PRIMARY KEY. Depends on sink. Doris does not support online PK change; the pipeline fails and requires manual sink recreation.

Exactly-once semantics — the mechanism.

• Flink checkpointing. Every N seconds (default 30), the Flink coordinator broadcasts a checkpoint barrier through the graph. Sources snapshot their offsets (Postgres LSN, MySQL GTID, Mongo resume token); operators snapshot their state; sinks flush any in-flight writes.
• Two-phase commit. The sink's checkpoint handler acts as the pre-commit of a 2PC transaction. On checkpoint completion, the sink commits the transaction (Doris StreamLoad transaction commit; Iceberg catalog snapshot commit).
• Restart from checkpoint. On job failure, Flink restarts from the last completed checkpoint. Sources rewind to their captured offsets; sinks resume from their committed transactions. No data is dropped; no data is duplicated (assuming the sink honours idempotent commits).
• The gotcha. The sink must expose a transactional or idempotent-by-key write path. Any sink that only supports at-least-once (e.g., raw HTTP POST) can only give at-least-once end-to-end, regardless of Flink's checkpointing.

Common interview probes on schema evolution + EOS.

• "How does Flink CDC apply ADD COLUMN to a Doris sink?" — DDL event → coordinator → sink MetadataApplier → ALTER TABLE.
• "What's the difference between lenient and strict schema behaviour?" — lenient auto-applies safe ops; strict fails on any DDL.
• "How does Flink CDC give exactly-once?" — Flink checkpoints + sink 2PC (or idempotent-by-key).
• "What sinks give true EOS?" — Doris, Iceberg, Paimon, Snowflake (via MERGE). Kafka gives EOS only within the same Kafka cluster (transactional producer).

Worked example — ADD COLUMN on MySQL → auto ALTER on Doris

Detailed explanation. The canonical demo — a source MySQL table gets an ALTER TABLE ADD COLUMN, the Flink CDC pipeline propagates it, and 30 seconds later the Doris target has the same column. Walk through the timeline, the DDL event flow, and the observation you'd make in the Flink UI.

• T=0. Source: ALTER TABLE orders.orders ADD COLUMN region VARCHAR(32).
• T=1s. MySQL binlog emits a QUERY event with the DDL SQL.
• T=1.2s. Flink CDC MySQL source parses the DDL, emits a SchemaChangeEvent(kind=ADD, column=region, type=VARCHAR(32)).
• T=1.5s. Pipeline coordinator receives the event, applies lenient policy, calls Doris MetadataApplier.applyAddColumn(...).
• T=3s. Doris runs ALTER TABLE orders_unified ADD COLUMN region VARCHAR(32); new rows now carry the column.

Question. Walk through the exact DDL sequence and demonstrate the auto-propagation, then show the query that verifies both sides have the same schema.

Input.

Component	Before	After
MySQL orders.orders columns	id, customer_id, amount, status	id, customer_id, amount, status, region
Doris orders_unified columns	id, customer_id, amount, status	id, customer_id, amount, status, region
Time to propagate	—	~2 seconds

Code.

-- Step 1 — source-side DDL
-- run on MySQL master
ALTER TABLE orders.orders
  ADD COLUMN region VARCHAR(32) DEFAULT NULL;

-- Verify the DDL is in the binlog
SHOW BINLOG EVENTS IN 'mysql-bin.000042' LIMIT 20;
-- ... look for the ALTER TABLE query event ...

# pipeline.yaml (unchanged) — the pipeline itself does not need editing
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 5400-5403
  server-time-zone: UTC
  scan.incremental.snapshot.enabled: true

sink:
  type: doris
  name: orders-doris
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}

pipeline:
  name: orders-mysql-to-doris
  parallelism: 4
  schema.change.behavior: lenient        # <— this is what auto-propagates ADD COLUMN
  execution.checkpointing.interval: 30s

-- Step 3 — verify Doris now has the new column
-- run on Doris
DESCRIBE orders_unified;
-- Field       | Type         | ...
-- id          | BIGINT       | ...
-- customer_id | BIGINT       | ...
-- amount      | DECIMAL(10,2)| ...
-- status      | VARCHAR(32)  | ...
-- region      | VARCHAR(32)  | ...     ← new column, auto-added

-- Step 4 — verify new inserts carry the column
SELECT id, region FROM orders_unified WHERE region IS NOT NULL LIMIT 5;

Step-by-step explanation.

1. The source DDL runs normally on MySQL — the DBA does not need to coordinate with the CDC team. The ALTER TABLE ... ADD COLUMN is written to the binlog as a QUERY event.
2. The Flink CDC MySQL source reads the binlog, sees the QUERY event, parses the SQL, and emits a SchemaChangeEvent(kind=ADD_COLUMN, ...) into the Flink graph.
3. The pipeline coordinator receives the event. Because schema.change.behavior: lenient, the coordinator forwards the event to the Doris sink's MetadataApplier.
4. The Doris sink calls stream_load with a schema-refresh request, which triggers an ALTER TABLE ADD COLUMN on the Doris side. The DDL is executed within Doris's internal DDL queue and takes a few seconds.
5. Once the sink DDL succeeds, the pipeline resumes writing data events. New rows carry the region column; historical rows already in Doris have NULL for region (the default for a new nullable column).

Output.

Time (s)	Event	Location
0.0	ALTER TABLE ADD COLUMN region	MySQL master
1.0	binlog QUERY event emitted	MySQL binlog
1.2	SchemaChangeEvent emitted	Flink CDC source
1.5	ADD COLUMN forwarded to sink	Pipeline coordinator
3.0	ALTER TABLE completes	Doris FE
3.5	Data writes resume with new column	Flink CDC sink

Rule of thumb. For any sink that supports online ADD COLUMN (Doris, StarRocks, Paimon, Iceberg, Snowflake), schema.change.behavior: lenient is the correct default. The DBA can run DDL on the source without pre-notifying the CDC team; the pipeline auto-propagates within seconds.

Worked example — the DROP COLUMN behaviour choice

Detailed explanation. DROP COLUMN is where the schema-evolution behaviour matrix actually earns its keep. Lenient mode does not physically drop the column on the sink — it stops writing to the column but preserves the existing data. This is deliberately non-symmetric: sinks are analytical warehouses; historical data has value; dropping a column loses that history irreversibly.

• Source. ALTER TABLE users DROP COLUMN legacy_flag.
• Sink under lenient. Column legacy_flag remains on the sink; new rows have NULL; existing rows keep their historical values.
• Sink under strict. Pipeline fails; DBA must manually adjust the sink schema before restart.
• Sink under evolve. Column is dropped on the sink; historical data is lost.

Question. Walk through the three behaviours for a DROP COLUMN event and recommend the default for an analytics ingest pipeline.

Input.

Config	Behaviour on DROP
lenient	column stays on sink; new writes NULL
strict	pipeline fails; manual intervention
evolve	column dropped on sink; history lost
ignore	DDL swallowed; sink unchanged

Code.

# pipeline.yaml — explicitly per-op schema behaviour
source:
  type: mysql
  name: users-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: users.users
  server-id: 5700-5701
  scan.incremental.snapshot.enabled: true

sink:
  type: iceberg
  name: users-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: analytics.users

pipeline:
  name: users-mysql-to-iceberg
  parallelism: 2
  # Explicit per-op policy — 3.x variants expose these knobs
  schema.change.behavior: lenient
  # optional: individual op overrides
  # schema.change.behavior.add-column: evolve
  # schema.change.behavior.drop-column: lenient
  execution.checkpointing.interval: 60s

-- Source event
-- MySQL master:
ALTER TABLE users DROP COLUMN legacy_flag;

-- Iceberg sink under lenient — verify column stays
SELECT column_name, is_nullable
FROM   iceberg_catalog.analytics.users_schema
WHERE  table_name = 'users';
-- ... includes legacy_flag as nullable ...

-- New writes have NULL for legacy_flag
SELECT id, legacy_flag
FROM   analytics.users
WHERE  updated_at > current_date;
-- Every row shows legacy_flag = NULL

Step-by-step explanation.

1. Under lenient (recommended default), a DROP COLUMN on the source becomes a "stop writing this column" instruction to the sink. The column definition stays; new rows have NULL; historical rows keep their values. This is the safest behaviour for analytics — you never lose history to a source DDL you can't reverse.
2. Under strict, the pipeline fails immediately. The DBA must run the equivalent ALTER TABLE DROP COLUMN on the sink manually, then restart the pipeline from savepoint. Use this when the sink schema is managed externally (dbt, Alembic) and you want CDC to never mutate it.
3. Under evolve (an aggressive mode in some 3.x variants), the sink executes the DROP COLUMN. All historical data for that column is permanently deleted. This is symmetric with the source but the correctness cost is often unacceptable.
4. Under ignore, the DDL event is silently swallowed. The sink continues with its old schema; the CDC job continues writing data events, but any subsequent INSERT that references the dropped column will fail (source-side query error). Rarely correct.
5. The recommendation for analytics: default to lenient, override to strict only if the sink schema is under contract with a downstream team. Never default to evolve unless you have a compliance requirement to physically delete columns on drop (GDPR right-to-forget).

Output.

Behaviour	DROP COLUMN result on sink	Historical data	New data
lenient (recommended)	column stays	preserved	NULL
strict	pipeline fails	untouched	not written
evolve	column dropped	lost	not written
ignore	column stays	preserved	continues to be written (until source query fails)

Rule of thumb. For analytics ingest, lenient is the correct default for DROP COLUMN. Preserving history against a source DDL you cannot reverse is worth the cost of a NULL-column tail on the sink table. Downstream queries can filter or ignore the NULL tail; downstream data cannot be resurrected once dropped.

Worked example — exactly-once via checkpoint + Doris transactional stream_load

Detailed explanation. The EOS mechanism has two components: Flink's checkpoint barrier (guaranteeing the source position is snapshotted) and the sink's transactional write (guaranteeing the write is atomic per checkpoint). Doris supports transactional StreamLoad — the Flink CDC Doris sink opens a stream_load transaction per checkpoint, appends data through the checkpoint interval, and commits on checkpoint completion. On failure between checkpoint N and N+1, Doris aborts the in-flight transaction; Flink restarts from checkpoint N; the source rewinds to the LSN at checkpoint N; no data is lost or duplicated.

• Flink side. execution.checkpointing.mode = EXACTLY_ONCE, execution.checkpointing.interval = 30s.
• Doris sink side. Two-phase commit — pre-commit on checkpoint snapshot; commit on checkpoint complete; abort on job failure.
• Failure model. Job crash → restart from last completed checkpoint → source rewinds → sink aborts in-flight transaction → data is exactly-once.

Question. Configure the pipeline for EOS end-to-end and walk through what happens on a task-manager crash between checkpoints.

Input.

Component	Setting
Flink checkpoint mode	EXACTLY_ONCE
Checkpoint interval	30 s
Doris sink	stream_load transactional
Failure scenario	task manager crash at T=15s (mid-checkpoint)

Code.

# pipeline.yaml — EOS configuration
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 5800-5803
  scan.incremental.snapshot.enabled: true

sink:
  type: doris
  name: orders-doris
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}
  # EOS-specific knobs
  sink.enable-2pc: true                     # two-phase commit on
  sink.buffer-size: 10000
  sink.buffer-count: 3
  sink.max-retries: 3

pipeline:
  name: orders-mysql-to-doris-eos
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 30s
  execution.checkpointing.mode: EXACTLY_ONCE
  execution.checkpointing.timeout: 60000
  execution.checkpointing.min-pause: 5000
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 5
  restart-strategy.fixed-delay.delay: 10s

Timeline of a failure and recovery
==================================

T = 0s   Checkpoint N completes.
         MySQL binlog position acked at GTID = ...:42.
         Doris StreamLoad transaction #17 committed.

T = 5s   Data events 43..80 flow through Flink CDC.
         Doris StreamLoad transaction #18 opened, receiving events.

T = 15s  Task manager crashes.
         Doris StreamLoad transaction #18 has not committed.
         Doris TC (transaction coordinator) times out and aborts #18.

T = 20s  Flink restart-strategy kicks in.
         New task manager starts.
         Flink restores from Checkpoint N.
         MySQL source rewinds to GTID ...:42.
         Doris sink opens new StreamLoad transaction #19.

T = 25s  Data events 43..80 replay through Flink CDC (from source).
         Doris StreamLoad #19 receives events 43..80.

T = 55s  Checkpoint N+1 completes.
         MySQL binlog acked at GTID ...:80.
         Doris StreamLoad transaction #19 committed.

Outcome: events 43..80 are in Doris exactly once.
         No duplicates (transaction #18 aborted).
         No losses (transaction #19 has them all).

Step-by-step explanation.

1. Flink's checkpointing model captures the state of every operator (source offset, transform state, sink pre-commit state) atomically as of the checkpoint barrier. Between checkpoints, all writes are "in flight" from the EOS perspective.
2. The Doris sink implements TwoPhaseCommitSinkFunction. On snapshotState (checkpoint barrier arriving), the sink pre-commits — it tells the Doris TC that the transaction is ready to commit but does not commit yet. The transaction ID is stored in the Flink state.
3. On notifyCheckpointComplete (checkpoint fully persisted), the sink commits — the Doris TC finalises the transaction, making the data visible to queries. Only after this do the events count as "delivered" from the EOS perspective.
4. On a task manager crash between snapshot and notify, the transaction remains in "pre-committed" state on Doris. Doris's TC times out and aborts it; the data is dropped from Doris. Flink restarts, rewinds the source, and re-emits the events into a new transaction.
5. The gotcha: the sink must support 2PC or idempotent-by-key writes. Doris StreamLoad supports 2PC as of recent versions; Iceberg's catalog-commit protocol is naturally 2PC; Snowflake requires MERGE-into-target keyed by primary key; Kafka has transactional producers.

Output.

Property	Guaranteed
Source offset resume on restart	yes (from checkpoint)
Sink transaction atomicity per checkpoint	yes (2PC)
End-to-end exactly-once	yes
Latency floor	checkpoint interval (30s)
Latency ceiling on failure	2× checkpoint interval + restart delay

Rule of thumb. For EOS, pick a sink that supports 2PC (Doris, Iceberg, Paimon), set execution.checkpointing.mode: EXACTLY_ONCE, and accept the 30-second data-visibility floor. Sinks without 2PC (raw HTTP, some older warehouses) can only give at-least-once end-to-end — pick idempotent-by-key writes on those to get effective EOS.

Senior interview question on schema evolution + EOS

A senior interviewer might ask: "Walk me through what happens end-to-end when a source MySQL table adds a column mid-stream, in a pipeline writing to Iceberg with EXACTLY_ONCE semantics. Cover the DDL event flow, the checkpoint interaction, the Iceberg catalog commit, and the failure story if the sink DDL times out."

Solution Using the DDL-flushes-before-data model + Iceberg catalog transactions

# pipeline.yaml — MySQL → Iceberg with schema evolution + EOS
source:
  type: mysql
  name: orders-mysql-src
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 5900-5903
  scan.incremental.snapshot.enabled: true

sink:
  type: iceberg
  name: orders-iceberg-sink
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: analytics.orders
  # EOS-specific
  sink.write.format.default: parquet
  sink.upsert.enabled: true
  sink.upsert.equality-column-names: id

pipeline:
  name: orders-mysql-to-iceberg-eos
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 60s        # Iceberg commits are slower — bigger interval
  execution.checkpointing.mode: EXACTLY_ONCE
  execution.checkpointing.timeout: 120000      # 2 minutes — Iceberg commits can be slow
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 5
  restart-strategy.fixed-delay.delay: 30s

End-to-end flow: ADD COLUMN mid-stream, with EOS
================================================

T = 0     Checkpoint N-1 complete. MySQL LSN acked. Iceberg snapshot committed.

T = 5s    Data events 43..80 in flight through Flink CDC.

T = 12s   Source: ALTER TABLE orders.orders ADD COLUMN region VARCHAR(32).
          Binlog emits QUERY event at position 81.

T = 12.5s Flink CDC source parses DDL, buffers pending data events (81+),
          emits SchemaChangeEvent to coordinator.

T = 13s   Coordinator forwards to Iceberg sink MetadataApplier.
          Iceberg sink issues:
              ALTER TABLE analytics.orders ADD COLUMN region STRING
          via the Hive Metastore catalog.

T = 15s   Iceberg catalog ADD COLUMN succeeds.
          Sink resumes writing; new events carry `region`.

T = 30s   Checkpoint N barrier flows through graph.
          Source snapshots LSN (post-DDL position).
          Sink pre-commits Iceberg snapshot files with new schema.

T = 32s   Checkpoint N fully committed:
          - Iceberg catalog snapshot references the new files.
          - New Iceberg schema is visible to query engines.

T = 33s   Post-checkpoint: pipeline continues.
          Downstream queries see new `region` column with data.

Step-by-step trace.

Question	Answer	Reasoning
What if the sink DDL times out?	pipeline fails; auto-restart	The checkpointing.timeout gives 2 min to complete DDL + commit
Are events between T=5s and T=12s lost?	no	Pre-DDL events flush before ADD COLUMN forwarded
What sees the new column first?	Iceberg catalog snapshot after checkpoint N	Schema visibility is atomic per snapshot
What if the source DDL is rolled back?	irreversible — sink already has the column	This is why lenient is safer than evolve

After the flow, the Iceberg table has the new column, historical rows have NULL for region, new rows have real values, and the atomic snapshot boundary means downstream queries either see the entirely-old schema or the entirely-new schema — never a mix.

Output:

Aspect	Result
Latency of DDL propagation	~3 seconds (source to sink)
First checkpoint after DDL	includes both pre-DDL and post-DDL data atomically
Downstream query consistency	atomic — one snapshot boundary
Failure recovery	full restart, source rewinds, sink 2PC aborts
Cost	one extra Iceberg snapshot commit per DDL event

Why this works — concept by concept:

• DDL flushes before data — the Flink CDC coordinator buffers post-DDL data events until the sink DDL completes. This guarantees the sink schema is ready before the first data event that uses it. No race conditions.
• Iceberg catalog snapshot atomicity — every checkpoint becomes an Iceberg snapshot; the schema change and the corresponding data files land in the same snapshot. Query engines see a coherent schema-and-data pair per snapshot.
• Two-phase commit at the sink — pre-commit at checkpoint snapshot, commit at checkpoint completion. Failure between snapshot and complete aborts the in-flight transaction; the source rewinds; no duplication.
• Lenient is the safe default — auto-propagates ADD, safe-handles DROP (preserves history), widens types compatibly, fails only on incompatible changes. The alternative modes (strict, evolve) are for specific compliance or contract cases.
• Cost — one extra Iceberg catalog commit per DDL event (~1 second). Schema evolution is not free but is proportional to the DDL rate, which is typically < 1 per hour per table. The steady-state EOS cost is checkpoint_interval / event_rate extra latency — 30-60 seconds is normal.

Streaming
Topic — streaming
Streaming EOS and checkpoint problems

Practice →

SQL Topic — sql SQL problems on schema evolution and DDL propagation

Practice →

---

5. Sink connectors + comparison to Debezium

Six sinks, one Flink job — Doris, StarRocks, Paimon, Snowflake, BigQuery, Iceberg

The mental model in one line: Flink CDC 3.x ships first-party sink connectors for Doris, StarRocks, Paimon, Iceberg, Snowflake, and BigQuery (plus Kafka for the fan-out pattern) — each with its own transactional-write mechanism and its own schema-evolution DDL API — and the sink choice determines the write latency floor, the query engine you can layer on top, and the migration cost from the legacy Debezium+Kafka stack. Choosing the right sink is 50% of the ingest architecture; the source connectors are relatively uniform, but the sinks vary dramatically in write model, cost, and analytical fit.

Sink family 1 — real-time OLAP.

• Doris. Sub-second query on billion-row tables; primary-key merge-on-read via unique key tables; StreamLoad transactional writes with 2PC. Best fit for interactive dashboards + CDC ingest.
• StarRocks. Fork of Doris (now separate project); similar write model (transactional stream_load); similar analytical strengths. Choose based on org / cloud provider fit.
• Paimon. Table format on top of object storage (S3 / OSS); LSM-tree write path with periodic compaction; deep Flink integration. Best for lakehouse patterns where the ingest table is also queried by Spark / Trino.

Sink family 2 — cloud data warehouse.

• Snowflake. Sink writes via COPY INTO from a staging table, with a MERGE-into-target for upsert semantics. Latency floor ~1 minute (COPY INTO cadence). Best fit when the analytics stack is already Snowflake.
• BigQuery. Sink writes via BigQuery Storage Write API with commit modes (pending, immediate). Latency floor ~30 seconds. Best fit for GCP-native stacks with existing BigQuery pipelines.

Sink family 3 — open table format.

• Iceberg. Table format on top of Parquet + a catalog (Hive Metastore, AWS Glue, Nessie). Sink writes Parquet files + commits a catalog snapshot per Flink checkpoint. Latency floor = checkpoint interval (30–60s). Best fit for multi-engine lakehouses (Spark + Trino + Snowflake external tables).
• Kafka (fan-out). Sink writes Debezium-JSON events to a Kafka topic. Not a terminal sink; typically feeds downstream Kafka consumers. Best fit for multi-consumer patterns.

Comparison to legacy Debezium + Kafka + sink.

• Ops surface. Legacy: Kafka Connect worker (Debezium) + Kafka broker + Flink cluster + sink writer. Flink CDC: Flink cluster + sink. Difference: 3 tiers vs 1 tier.
• Latency. Legacy: source → Debezium → Kafka → Flink → sink adds ~300 ms. Flink CDC: source → Flink → sink adds ~50 ms.
• Schema evolution. Legacy: 4-team coordination on ADD COLUMN. Flink CDC: auto-propagate via lenient.
• Checkpointing. Legacy: 3 stories (Debezium offset commits, Kafka retention, Flink checkpoints). Flink CDC: 1 story (Flink checkpoints).
• When to still use Kafka. Multi-consumer fan-out; cross-team schema contract; > 12h replay window.

Migration path — legacy stack to Flink CDC 3.x.

• Phase 1. Pick one low-risk pipeline (audit log). Stand up parallel Flink CDC pipeline; write to shadow sink table.
• Phase 2. Reconcile — row-level diff between legacy sink and shadow sink. Zero rows implies parity.
• Phase 3. Cut over reads to shadow sink; decommission legacy pipeline for that source.
• Phase 4. Repeat for the next 5 pipelines; batch cutover per source database.
• Phase 5. Decommission Kafka + Kafka Connect once all pipelines migrated (or scope Kafka to the fan-out patterns that still need it).

Common interview probes on sinks.

• "Which sink gives sub-second query latency on CDC-ingested data?" — Doris or StarRocks (unique-key tables).
• "How does the Iceberg sink handle schema evolution?" — new snapshot per checkpoint; catalog ADD COLUMN atomic with data.
• "What is the Snowflake sink's latency floor?" — ~1 minute (COPY INTO cadence).
• "When is Kafka still the right sink?" — multi-consumer fan-out; replay window > source WAL retention.

Worked example — Doris sink config for real-time OLAP

Detailed explanation. Doris is the most common CDC sink for interactive dashboards in 2026 — unique-key tables give MERGE-on-read semantics natively, StreamLoad is fully transactional (2PC-compatible), and the DDL API supports online ADD COLUMN. Walk through the sink config, the target table's DDL, and the query pattern that hits sub-second latencies.

• Target table. orders_unified — unique key on id, replication = 3, aggregation = REPLACE.
• Sink knobs. sink.enable-2pc: true, sink.enable-delete: true, sink.buffer-size: 10000.
• Latency floor. ~1 second from source commit to Doris visibility (dominated by StreamLoad batching).

Question. Configure a Doris sink for a high-throughput orders ingest (~10k events/sec) with 2PC EOS and sub-second query latency.

Input.

Parameter	Value
Source rate	10k events/sec
Doris FE endpoint	doris-fe.internal:8030
Target database	orders_analytics
Target table	orders_unified
Query SLA	p99 < 500 ms

Code.

-- Doris side — create the target table
CREATE TABLE orders_analytics.orders_unified (
    id           BIGINT       NOT NULL,
    customer_id  BIGINT       NOT NULL,
    product_id   BIGINT       NOT NULL,
    quantity     INT          NOT NULL,
    price        DECIMAL(10,2) NOT NULL,
    status       VARCHAR(32)  NOT NULL,
    created_at   DATETIME     NOT NULL,
    updated_at   DATETIME     NOT NULL,
    shard_id     TINYINT      NOT NULL
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 32
PROPERTIES (
    "replication_num"       = "3",
    "enable_unique_key_merge_on_write" = "true",
    "compression"           = "LZ4",
    "storage_medium"        = "SSD"
);

# pipeline.yaml — Flink CDC MySQL → Doris
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 6000-6003
  scan.incremental.snapshot.enabled: true

sink:
  type: doris
  name: orders-doris
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}
  # Target
  table.identifier: orders_analytics.orders_unified
  # 2PC EOS
  sink.enable-2pc: true
  sink.enable-delete: true
  # Batching for 10k events/sec throughput
  sink.buffer-size: 10485760          # 10 MB per micro-batch
  sink.buffer-count: 3                 # 3 buffers in flight
  sink.max-retries: 3
  # Stream load properties
  sink.properties.format: json
  sink.properties.strip_outer_array: true
  sink.properties.read_json_by_line: true
  sink.properties.exec_mem_limit: 2147483648   # 2 GB per StreamLoad

pipeline:
  name: orders-mysql-to-doris-realtime
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 10s        # 10s for sub-second SLA
  execution.checkpointing.mode: EXACTLY_ONCE

-- Query pattern that hits sub-second latency
SELECT customer_id,
       count(*)         AS orders,
       sum(price)       AS total_spend,
       max(updated_at)  AS last_order
FROM   orders_analytics.orders_unified
WHERE  status = 'PAID'
  AND  updated_at > now() - INTERVAL 1 DAY
GROUP  BY customer_id
ORDER  BY total_spend DESC
LIMIT  100;

Step-by-step explanation.

1. The Doris target table uses UNIQUE KEY(id) — every row is uniquely identified by id, and updates on the same id are merged. This is the natural fit for CDC: an UPDATE on the source becomes an upsert on Doris; a DELETE becomes a tombstone.
2. enable_unique_key_merge_on_write = true uses Doris's MoW (merge-on-write) engine — writes are more expensive but reads are faster because no merge is needed at query time. For a query-heavy workload, this is the right trade-off.
3. sink.enable-2pc: true on the Flink side pairs with Doris's transactional StreamLoad. Every checkpoint interval (10s in this config), Flink pre-commits a StreamLoad transaction; on checkpoint completion, the transaction is fully committed. Data visibility is atomic per checkpoint.
4. sink.buffer-size: 10 MB gives Doris large micro-batches, which amortises the StreamLoad HTTP overhead. At 10k events/sec, a 10 MB buffer fills in ~5 seconds — well within the 10s checkpoint interval.
5. The query pattern shows sub-second latency because Doris's columnar storage + primary-key routing + MoW engine keeps hot data in memory. p99 < 500 ms is achievable for ~1000 rpm at 100M row scale.

Output.

Metric	Value
Source rate	10k events/sec
Sink checkpoint interval	10 s
Data visibility latency	~10 s (checkpoint-bound)
Query p99 latency	< 500 ms
EOS end-to-end	yes (2PC)
Storage per 1M rows	~200 MB (LZ4 compressed)

Rule of thumb. Doris with unique-key MoW + 2PC StreamLoad is the 2026 default for CDC + interactive dashboard workloads. Set the Flink checkpoint interval to match your data-visibility SLA — 10s for interactive, 30s for near-real-time, 60s for batch-like.

Worked example — Iceberg sink for open-lakehouse pattern

Detailed explanation. Iceberg is the sink of choice when the ingest table is queried by multiple engines — Spark for batch, Trino for interactive, Snowflake for external-table joins. The Flink CDC Iceberg sink writes Parquet files and commits a catalog snapshot per Flink checkpoint. The trade-off: latency floor of one checkpoint (30–60s), but multi-engine reads and full Iceberg feature set (time-travel, hidden partitioning, schema evolution, row-level updates).

• Catalog. Hive Metastore, AWS Glue, Nessie, or REST.
• Storage. S3 / GCS / OSS (Parquet files).
• Write model. Copy-on-write (default) or merge-on-read (delete files).

Question. Configure an Iceberg sink for a CDC pipeline with upsert semantics (source has UPDATEs), and demonstrate the multi-engine query story.

Input.

Parameter	Value
Iceberg catalog	Hive Metastore
Warehouse	s3://analytics-lake/warehouse
Target table	analytics.orders
Upsert equality column	id
Query engines	Spark + Trino + Snowflake external

Code.

# pipeline.yaml — MySQL → Iceberg with upsert
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 6100-6103
  scan.incremental.snapshot.enabled: true

sink:
  type: iceberg
  name: orders-iceberg
  # Catalog config
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  # Target
  table.identifier: analytics.orders
  # Upsert semantics
  sink.upsert.enabled: true
  sink.upsert.equality-column-names: id
  # File format + compression
  sink.write.format.default: parquet
  sink.write.parquet.compression-codec: zstd
  sink.write.parquet.row-group-size-bytes: 134217728    # 128 MB
  # File compaction (async)
  sink.commit.max-committed-checkpoint-count: 100

pipeline:
  name: orders-mysql-to-iceberg-upsert
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 60s
  execution.checkpointing.mode: EXACTLY_ONCE

-- Spark side — batch query joining CDC-ingested data with dimension tables
SELECT o.id,
       o.customer_id,
       o.amount,
       c.name         AS customer_name,
       c.tier         AS customer_tier
FROM   iceberg.analytics.orders  o
JOIN   iceberg.analytics.customers c
       ON o.customer_id = c.id
WHERE  o.updated_at > current_date - INTERVAL 7 DAYS
  AND  o.status = 'PAID';

-- Trino side — interactive query for a dashboard
SELECT date_trunc('hour', updated_at) AS hour,
       count(*)                       AS orders,
       sum(amount)                    AS revenue
FROM   iceberg.analytics.orders
WHERE  updated_at > current_timestamp - INTERVAL '1' DAY
GROUP  BY 1
ORDER  BY 1 DESC;

-- Snowflake side — external table over the Iceberg warehouse
CREATE OR REPLACE EXTERNAL TABLE ext_orders
  WITH LOCATION       = @analytics_lake_stage/warehouse/analytics/orders
       FILE_FORMAT    = (TYPE = PARQUET)
       AUTO_REFRESH   = TRUE
       TABLE_FORMAT   = 'ICEBERG';

SELECT id, customer_id, amount, status
FROM   ext_orders
WHERE  updated_at > current_date - 7;

Step-by-step explanation.

1. The Iceberg sink is configured with sink.upsert.enabled: true and sink.upsert.equality-column-names: id. Iceberg's V2 spec supports row-level upserts via delete files (positional deletes for row-level replacement). An UPDATE on the source becomes a delete-file entry (for the old row) + a data-file entry (for the new row).
2. Per Flink checkpoint, the sink writes Parquet files (data + deletes) and commits an Iceberg snapshot via the Hive Metastore catalog. The snapshot is atomic: readers see the pre-commit state or the post-commit state, never a mix.
3. sink.commit.max-committed-checkpoint-count: 100 caps the number of retained snapshots — older snapshots are eligible for cleanup by a separate compaction job. Without this, the metadata grows unboundedly.
4. Spark and Trino both read Iceberg natively; the Snowflake external-table feature reads Iceberg tables directly from the catalog. Three query engines, one ingest pipeline, one storage layer.
5. The 60-second checkpoint interval is chosen for Iceberg's commit overhead — HMS catalog commits are seconds, and pushing checkpoints too frequently generates too many small snapshots. For interactive dashboards, layer Doris on top and use Iceberg for batch + external-tool queries.

Output.

Query engine	Latency (last-hour)	Latency (full-history batch)	Notes
Doris (real-time layer)	< 500 ms	N/A	Separate sink
Trino (interactive)	1–3 s	30–60 s	Native Iceberg read
Spark (batch)	N/A	5–15 min	Batch job over 100 GB
Snowflake (external)	5–20 s	1–5 min	External-table cache warms

Rule of thumb. Use Iceberg as the storage of record; layer Doris (or StarRocks) on top for sub-second queries; use Spark for heavy transformations. The CDC pipeline writes once (to Iceberg); the sub-second layer is a downstream materialisation.

Worked example — the migration from Debezium+Kafka+Flink to Flink CDC 3.x

Detailed explanation. The migration story is the practical use case for most senior interviews on this topic. A team runs a legacy Debezium + Kafka + Flink pipeline for a MySQL → Snowflake ingest. The migration target is Flink CDC 3.x direct (no Kafka in the middle, one Flink job). Walk through the parallel-run pattern, the reconciliation query, and the cutover.

• Legacy stack. Debezium reads MySQL binlog → Kafka topic dbserver1.orders.orders → Flink Java job → Snowflake COPY INTO.
• Target stack. Flink CDC MySQL source → Snowflake sink (COPY INTO with MERGE).
• Parallel-run period. 2 weeks with row-level reconciliation.

Question. Design the parallel-run architecture, the reconciliation query, and the cutover runbook.

Input.

Component	Legacy	New
Source	MySQL binlog	MySQL binlog (Flink CDC)
Middle	Kafka + Debezium	(none)
Sink	Snowflake orders_legacy	Snowflake orders_new
Reconciliation	daily diff	daily diff
Cutover	after 2 weeks parity	after 2 weeks parity

Code.

# pipeline-new.yaml — the Flink CDC 3.x replacement
source:
  type: mysql
  name: orders-mysql
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 6200-6203
  scan.incremental.snapshot.enabled: true

sink:
  type: snowflake
  name: orders-snowflake-new
  jdbc.url: jdbc:snowflake://account.snowflakecomputing.com
  username: cdc_writer
  password: ${SNOWFLAKE_PASSWORD}
  database: analytics
  schema: cdc
  # Target — shadow table during parallel run
  table.identifier: orders_new
  # Snowflake write model
  sink.write.mode: merge
  sink.merge.equality-column-names: id
  sink.stage.name: cdc_stage
  sink.stage.format: parquet

pipeline:
  name: orders-mysql-to-snowflake-new
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 60s
  execution.checkpointing.mode: EXACTLY_ONCE

-- Reconciliation query — run daily during the 2-week parallel run
-- Compare orders_legacy (Debezium + Kafka + Flink) to orders_new (Flink CDC)

-- Row counts
SELECT 'legacy' AS pipeline, COUNT(*) AS n
FROM   analytics.cdc.orders_legacy
WHERE  updated_at::date = current_date - 1
UNION ALL
SELECT 'new'    AS pipeline, COUNT(*) AS n
FROM   analytics.cdc.orders_new
WHERE  updated_at::date = current_date - 1;
-- Expect: n_legacy = n_new

-- Row-level diff — the acceptance test
WITH legacy AS (
  SELECT id, customer_id, amount, status,
         md5(concat(id, '|', customer_id, '|', amount, '|', status)) AS row_hash
  FROM   analytics.cdc.orders_legacy
  WHERE  updated_at::date = current_date - 1
),
new AS (
  SELECT id, customer_id, amount, status,
         md5(concat(id, '|', customer_id, '|', amount, '|', status)) AS row_hash
  FROM   analytics.cdc.orders_new
  WHERE  updated_at::date = current_date - 1
)
SELECT 'only_in_legacy' AS side, id FROM legacy
  WHERE NOT EXISTS (SELECT 1 FROM new WHERE new.id = legacy.id
                     AND new.row_hash = legacy.row_hash)
UNION ALL
SELECT 'only_in_new'    AS side, id FROM new
  WHERE NOT EXISTS (SELECT 1 FROM legacy WHERE legacy.id = new.id
                     AND legacy.row_hash = new.row_hash);
-- Acceptance: 0 rows for 14 consecutive days.

Cutover runbook — Flink CDC 3.x migration
=========================================

Day 0
   Deploy pipeline-new.yaml alongside legacy stack.
   Both pipelines writing to Snowflake (different tables).
   Set up daily reconciliation query.

Days 1..14
   Daily reconciliation must return 0 diverging rows.
   Any divergence triggers a P2 investigation; parallel run extended.

Day 15 (assuming 14 days of parity)
   Announce cutover window (T minus 24h)
   Freeze legacy pipeline PRs.
   Prepare downstream dashboard rename.

Day 16 (cutover)
   T-0    Pause the legacy Flink job.
   T+1m   Verify Snowflake orders_legacy has final events (from Kafka drain).
   T+5m   Rename Snowflake orders_new → orders (production).
   T+5m   Rename orders_legacy → orders_legacy_frozen.
   T+10m  Update downstream views to point at orders.
   T+15m  Run smoke tests on production dashboards.
   T+30m  If green, tear down Kafka topic + Debezium worker.
   T+2h   Decommission legacy Flink job.

Rollback:
   If T+30m smoke tests fail, rename back:
     orders → orders_new
     orders_legacy_frozen → orders_legacy
     Point downstream views at orders_legacy.
     Resume legacy Flink job from Kafka offset held during freeze.

Step-by-step explanation.

1. The parallel-run phase (days 1–14) is non-negotiable. Both pipelines write to separate Snowflake tables (orders_legacy and orders_new); the reconciliation query proves row-level parity. Any divergence blocks cutover.
2. The reconciliation uses a md5(concat(...)) row hash for a fast row-level comparison. A full outer semi-join with anti-matches identifies rows present in only one side. Zero rows for 14 consecutive days is the acceptance bar.
3. The cutover is a table-rename operation. Snowflake ALTER TABLE ... RENAME TO is metadata-only and near-instant; downstream views need a one-shot pointer update.
4. The rollback plan is symmetric — reverse-rename the tables, point views back at orders_legacy, resume the legacy Flink job from its held Kafka offset. This is possible only because Kafka retention held the offset during the freeze; the legacy job can catch up.
5. The teardown (T+30m onward) is the last irreversible step. Once Kafka + Debezium are decommissioned, rollback becomes a rebuild. Delay the teardown until the smoke tests have been green for at least an hour.

Output.

Phase	Duration	Success criterion
Deploy new pipeline	1 day	New pipeline healthy
Parallel run + reconciliation	14 days	0 diverging rows daily
Cutover window	30 minutes	Smoke tests green
Teardown legacy stack	2 hours	No downstream errors

Rule of thumb. Every CDC migration is a 3-week story — 1 day of deploy, 14 days of parallel run + reconciliation, 1 day of cutover. Compressing this timeline is how migrations turn into outages. Row-level reconciliation is not optional; count-based reconciliation misses subtle divergences that only surface after cutover.

Senior interview question on sink selection + migration

A senior interviewer might ask: "You're choosing the sink for a new CDC pipeline serving an interactive dashboard (sub-second query SLA) plus a batch reporting layer (daily). Walk me through the sink options, the trade-offs, and the specific architecture you'd ship — including whether you'd need multiple sinks and how you'd handle schema evolution across them."

Solution Using a two-sink architecture — Doris for real-time, Iceberg for batch

# pipeline-doris.yaml — real-time layer
source:
  type: mysql
  name: orders-mysql-doris
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 6300-6303
  scan.incremental.snapshot.enabled: true

sink:
  type: doris
  name: orders-doris
  fenodes: doris-fe.internal:8030
  username: doris_writer
  password: ${DORIS_PASSWORD}
  table.identifier: orders_realtime.orders
  sink.enable-2pc: true
  sink.enable-delete: true

pipeline:
  name: orders-mysql-to-doris-realtime
  parallelism: 4
  schema.change.behavior: lenient
  execution.checkpointing.interval: 10s
  execution.checkpointing.mode: EXACTLY_ONCE

# pipeline-iceberg.yaml — batch layer
source:
  type: mysql
  name: orders-mysql-iceberg
  hostname: mysql.internal
  port: 3306
  username: flink_cdc
  password: ${MYSQL_CDC_PASSWORD}
  tables: orders.orders
  server-id: 6304-6307
  scan.incremental.snapshot.enabled: true

sink:
  type: iceberg
  name: orders-iceberg
  catalog.type: hive
  catalog.uri: thrift://hms.internal:9083
  warehouse: s3://analytics-lake/warehouse
  table.identifier: analytics.orders
  sink.upsert.enabled: true
  sink.upsert.equality-column-names: id
  sink.write.format.default: parquet
  sink.write.parquet.compression-codec: zstd

pipeline:
  name: orders-mysql-to-iceberg-batch
  parallelism: 2
  schema.change.behavior: lenient
  execution.checkpointing.interval: 60s
  execution.checkpointing.mode: EXACTLY_ONCE

Step-by-step trace.

Aspect	Doris pipeline	Iceberg pipeline
Sink	Doris (real-time)	Iceberg (batch)
Latency floor	10 s	60 s
Query engine	Doris SQL	Trino / Spark / Snowflake ext
Query SLA	sub-second	5–30 s (Trino), 5–15 min (Spark)
Checkpoint interval	10 s	60 s
Server-id	6300-6303	6304-6307
Schema evolution	auto (lenient)	auto (lenient)

After deploy, both pipelines run as independent Flink jobs on the shared cluster. The interactive dashboard queries Doris; the batch reports query Iceberg. Schema evolution on the MySQL source propagates to both sinks independently and atomically. If one pipeline fails, the other continues; the two are fully decoupled beyond the shared MySQL source.

Output:

Layer	Sink	Latency	Query engine	Use case
Real-time	Doris	10 s freshness	Doris SQL	Interactive dashboard
Batch	Iceberg	60 s freshness	Trino / Spark / Snowflake ext	Reports, BI
Shared source	MySQL binlog	—	—	Two independent binlog readers
Total infra	2 Flink jobs + 1 Doris cluster + 1 Iceberg catalog	—	—	Shared Flink cluster

Why this works — concept by concept:

• Two sinks, one source — each pipeline opens its own MySQL binlog reader with its own server-id. The pipelines are fully independent; failure of one does not affect the other. This is the correct pattern for multi-sink workloads.
• Match the sink to the query SLA — Doris for sub-second interactive; Iceberg for multi-engine batch. Using Iceberg for the interactive dashboard would miss the SLA; using Doris for the batch layer would lock in a single query engine.
• Independent checkpoint intervals — the Doris pipeline checkpoints every 10 s to meet the freshness SLA; the Iceberg pipeline checkpoints every 60 s to amortise the catalog commit cost. Each sink dictates its own checkpoint cadence.
• Schema evolution to both sinks — lenient on both pipelines means an ADD COLUMN on the source propagates to Doris and Iceberg independently. No coordination between the two sinks.
• Cost — two Flink jobs (~6 task slots each = 12 slots total) + Doris cluster (~8 vCPU) + Iceberg storage (~$0.023 / GB / month S3). The two-sink cost is roughly 1.5× the single-sink cost — well worth it for the different query SLAs.

ETL
Topic — etl
ETL sink connector and lakehouse ingest problems

Practice →

Optimization
Topic — optimization
Optimization problems on ingest throughput and sink write latency

Practice →

---

Cheat sheet — Flink CDC recipes

• When to use Flink CDC 3.x. Default choice for new ingest pipelines in 2026 where the source is Postgres / MySQL / Mongo / Oracle and the sink is Doris / StarRocks / Paimon / Iceberg / Snowflake / BigQuery. Keep the legacy Debezium + Kafka + Flink pattern only when multi-consumer fan-out or > 12h replay window is required.
• Pipeline YAML skeleton. source: (type, credentials, tables, server-id, snapshot mode) + optional transform: (projection, filter, derived columns) + optional route: (table remapping) + sink: (type, credentials, target) + pipeline: (name, parallelism, schema.change.behavior, checkpointing.interval + mode). Full pipeline in ~40 lines.
• MySQL source config. type: mysql, server-id: <unique-range>, server-time-zone: UTC, scan.incremental.snapshot.enabled: true, scan.startup.mode: initial, GTID + row-format binlog required on the server side, flink_cdc user with SELECT + RELOAD + REPLICATION SLAVE + REPLICATION CLIENT.
• Postgres source config. type: postgres, slot.name, decoding.plugin.name: pgoutput, publication.name, server-side wal_level = logical, max_replication_slots >= 5, dedicated CDC role with REPLICATION + SELECT only. Always ship the slot-lag alert.
• MongoDB source config. type: mongodb, hosts: <replica-set>, connection.options: replicaSet=rs0&readPreference=secondaryPreferred, copy.existing: true, heartbeat.interval.ms: 30000. Always know the oplog window; alert at 25% of the window.
• Schema evolution behaviour matrix. lenient (default, recommended for analytics) — auto ADD, preserve DROP, widening ALTER, safe RENAME. strict — fail on any DDL, use when sink schema is externally managed. ignore — swallow DDL, rarely correct. evolve — aggressive DROP that loses history, use only for compliance.
• EOS configuration. execution.checkpointing.mode: EXACTLY_ONCE, execution.checkpointing.interval: 30s (or 10s for interactive SLA, 60s for slow-committer sinks), sink must support 2PC (Doris, Iceberg, Paimon) or idempotent-by-key upsert (Snowflake MERGE).
• Doris sink config. type: doris, fenodes: <fe-endpoint>, sink.enable-2pc: true, sink.enable-delete: true, sink.buffer-size: 10485760, target table uses UNIQUE KEY(id) with enable_unique_key_merge_on_write = true. Latency floor ~10s; sub-second query.
• Iceberg sink config. type: iceberg, catalog.type: hive, catalog.uri, warehouse, table.identifier, sink.upsert.enabled: true with sink.upsert.equality-column-names: id, sink.write.format.default: parquet, sink.write.parquet.compression-codec: zstd. Latency floor = checkpoint interval; multi-engine reads.
• Snowflake sink config. type: snowflake, jdbc.url, database + schema + table.identifier, sink.write.mode: merge, sink.merge.equality-column-names: id, sink.stage.name, sink.stage.format: parquet. Latency floor ~1 minute; MERGE-into-target for upsert EOS.
• Kafka sink for fan-out. type: kafka, properties.bootstrap.servers, topic, value.format: debezium-json, properties.compression.type: zstd. Downstream consumers use Flink CDC Kafka source. Use only for multi-consumer or replay-window patterns.
• Savepoint restart. Trigger with Flink CLI: flink savepoint <job-id> s3://savepoints/. Restart with: flink run -s s3://savepoints/<savepoint-id> pipeline.yaml. Savepoints carry the source position (LSN / GTID / resume token) plus sink pre-commit state. Always take a savepoint before pipeline upgrades.
• Parallelism sizing. Set parallelism in the pipeline: header to 2–4× the number of source tables for the snapshot phase (Flink CDC parallelises the initial backfill by table + chunk). For steady-state streaming, 1–2× the number of source tables is sufficient.
• Migration path from legacy Debezium + Kafka + Flink. Phase 1: pilot pipeline in parallel (2 weeks). Phase 2: daily row-level reconciliation (0 diverging rows required). Phase 3: cutover reads (table rename). Phase 4: teardown legacy Kafka + Kafka Connect. Total per-pipeline effort: ~3 engineer-months; payback within 4 months of cutover for 20+ pipelines.
• Slot / binlog / oplog hygiene. Every pipeline retirement must drop the Postgres replication slot, confirm MySQL binlog retention > pipeline downtime, and monitor MongoDB oplog window vs CDC lag. Skipping the retirement step is the #1 CDC production incident.

Frequently asked questions

What is Flink CDC?

Flink CDC (short for Flink Change Data Capture) is a family of Apache Flink source and sink connectors packaged with a declarative YAML pipeline DSL that lets you build streaming ingest from Postgres / MySQL / MongoDB / Oracle sources to Doris / StarRocks / Paimon / Iceberg / Snowflake / BigQuery sinks as a single Flink job — no Kafka in the middle. Under the hood the source connectors wrap the Debezium engine as an embedded library (not as a separate Kafka Connect worker) so the log-tap protocols are the same battle-tested ones (pgoutput, MySQL binlog, Mongo change streams, Oracle LogMiner / XStream). The Flink CDC 3.x release introduced the pipeline YAML DSL and full end-to-end schema-evolution auto-propagation, which is what made it the 2026 default for new ingest pipelines. The name "Flink CDC" refers to the connector family + the YAML pipeline runtime; the underlying job is a normal Flink job that any Flink cluster can execute.

Flink CDC vs Debezium — do I still need Debezium?

You still get Debezium, but you no longer run it as a separate service. Flink CDC 3.x embeds the Debezium engine inside the Flink CDC source connector — the same Debezium codebase that parses MySQL binlog and Postgres logical replication, but running as a library inside your Flink task manager rather than as a Kafka Connect worker. What disappears is the Kafka Connect JVM, the Kafka broker (unless you keep it for fan-out), and the four-team coordination on schema evolution. What remains is the Debezium log-tap protocol and the wire-format compatibility with the Debezium-JSON schema. If you have a legacy Debezium + Kafka + Flink pipeline, migrating to Flink CDC 3.x is not a replacement of Debezium — it's a repackaging into a single-JVM ingest job that reduces the operational footprint by roughly 3×.

Do I still need Kafka with Flink CDC 3.x?

Only for three specific patterns. First, multi-consumer fan-out — two or more downstream consumers each want the change stream (analytics warehouse + search index + audit archiver). Kafka's log storage decouples the producer from N consumers. Second, cross-team schema contract — team A owns the source and teams B, C, D own the sinks; Kafka is the boundary where the Avro / JSON schema is defined (via Confluent Schema Registry or Apicurio). Third, long replay window — you need to be able to re-materialise the sink from N days ago because the sink logic changed; Kafka retention gives you a bounded replay window that the source WAL / binlog does not (they expire quickly). For every other pattern — single-consumer, single-team, no replay-beyond-source-retention — the single-Flink-job pattern is the correct default. The legacy three-hop stack (Debezium → Kafka → Flink) was a workaround for the pre-3.x world where Flink could not embed Debezium; that workaround is no longer required.

What is the Flink CDC pipeline YAML?

A declarative 40-line YAML file with four functional sections plus a header. source: names the source connector (type: mysql | postgres | mongodb | oracle), the credentials, and the tables to replicate. transform: (optional) applies per-row projection, filter, and derived-column expressions using a bounded SQL-like grammar — not full Flink SQL, no UDFs, no joins. route: (optional) remaps source tables to sink tables — commonly used for many-to-one merges or table renames. sink: names the sink connector (type: doris | iceberg | paimon | snowflake | bigquery | kafka), the credentials, and the target catalog / database / table. pipeline: is the header — job name, parallelism, schema.change.behavior policy (lenient is the recommended default), and checkpointing interval + mode. Together these five sections replace the ~500 lines of Java DataStream code that the equivalent pre-3.x pipeline required. The YAML is a facade over Flink DataStream; the runtime is a normal Flink job, so any cluster capacity, savepoint tooling, and monitoring pipeline that already exists for Flink continues to work.

Which sinks are supported by Flink CDC?

Six main sink families as of 2026. Real-time OLAP: Doris and StarRocks — sub-second query on billion-row tables via primary-key merge-on-read tables; StreamLoad transactional writes with 2PC EOS. Lakehouse: Paimon and Iceberg — Parquet files + a catalog snapshot per Flink checkpoint; multi-engine reads (Spark, Trino, Snowflake external); 30–60-second freshness floor. Cloud warehouse: Snowflake (COPY INTO + MERGE for upsert semantics; ~1-minute freshness) and BigQuery (Storage Write API with commit modes; ~30-second freshness). Fan-out: Kafka — for multi-consumer or replay-window patterns; write Debezium-JSON events into a Kafka topic and downstream consumers use the Flink CDC Kafka source to fan out. Each sink implements the Flink SinkV2 API with 2PC where the target system supports it; sinks without native 2PC (some older warehouses, raw HTTP endpoints) fall back to idempotent-by-primary-key upsert semantics for effective EOS. The sink choice sets your write latency floor, your query engine, and your migration cost from legacy stacks; it is 50% of the ingest architecture.

Does Flink CDC give exactly-once semantics?

Yes, when configured correctly and paired with a 2PC-capable sink. The mechanism is Flink's checkpoint model plus a two-phase-commit sink. On every checkpoint interval (default 30 seconds, tunable), Flink broadcasts a checkpoint barrier through the graph; each operator snapshots its state (source LSN / GTID / resume token, transform state, sink pre-commit state) atomically. When the checkpoint fully persists, the sink completes the second phase — Doris commits the StreamLoad transaction, Iceberg publishes the catalog snapshot, Snowflake commits the MERGE. On failure between checkpoints, Flink restarts from the last completed checkpoint; the source rewinds to the captured offset; the sink aborts any in-flight transaction. Data is neither dropped nor duplicated. The critical config: execution.checkpointing.mode: EXACTLY_ONCE, plus a checkpoint interval matching your SLA (10s for sub-second freshness, 60s for slow-committer sinks). The sink must support 2PC (Doris, Iceberg, Paimon are native; Snowflake via MERGE-by-PK; Kafka via transactional producer). Sinks that only support at-least-once writes (some HTTP endpoints, older warehouses) give effective EOS only if the sink write is idempotent by primary key — an upsert or MERGE that produces the same result under replay.

Practice on PipeCode

• Drill the streaming practice library → for the CDC design, checkpointing, and schema-evolution problems senior interviewers love.
• Rehearse on the ETL practice library → for the pipeline-topology, source-connector, and sink-migration problems that motivate Flink CDC in the first place.
• Sharpen the tuning axis with the optimization practice library → for the throughput-sizing, EOS-vs-at-least-once, and checkpoint-cadence problems.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the source-connector + sink-selection intuition against real graded inputs.

Lock in Flink CDC muscle memory

Docs explain YAML syntax. PipeCode drills explain the decision — when Flink CDC 3.x replaces Debezium+Kafka+Flink, when the pipeline YAML runs out of expressive power, when lenient schema evolution turns into a footgun. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice streaming problems →
Practice ETL problems →