I Built a CDC-to-Flink Pipeline and Documented Every Decision That Had to Change

8 architecture decisions, 5 Debezium incidents, 3 corrections — and what a streaming pricing engine looks like after all of them

The entire project was built with spec-driven development — every phase has a written spec before any code, every architecture decision has an ADR, and every incident has a root-cause document. The repository contains 8 ADRs, phase specs, event schema contracts, and an error-handling directory with 5 documented incidents. Three of those ADRs correct earlier ones. The Flink connector the architecture assumed existed turned out not to exist in Flink 2.0. A schema designed for clean JSON turned out to be impossible for Debezium to produce.

The running pipeline is the proof it works. The spec and ADR trail is the proof someone actually built it.

The pipeline

A property management company runs ~100 vacation rentals and needs dynamic pricing that accounts for real operational costs — not just market rates. The engine ingests two streams: operational costs via CDC (PostgreSQL → Debezium → Kafka) and market reference prices via Kinesis. Flink joins both, applies a formula that guarantees a minimum margin above cost, and writes pricing decisions to DynamoDB. Iceberg gets the audit trail downstream — not from Flink directly.

That’s the clean version. Here’s what it took to get there.

Five Debezium defaults that will break your pipeline silently

Every one of these incidents shared the same trait: the connector reported RUNNING while doing the wrong thing.

• Default key serialization. Debezium wraps the primary key in a JSON struct by default. Kafka partition assignment and Flink’s keyed state both require byte-identical keys for the same logical entity. A struct-wrapped apartment_id and a bare apartment_id produce different bytes. Partition affinity broke — events for the same apartment landed on different partitions, processing out of order. This ran undetected for weeks until an acceptance criterion verified it structurally (group-by-key and count per partition), not by spot-checking individual messages.
• Default topic naming. Debezium names topics <server>.<schema>.<table>. The Flink consumer expected a different topic name. Both services ran. Both reported healthy. Zero data flowed between them. Diagnosing this required checking the data plane (count messages on the actual topic), not the control plane (connector status endpoint).
• Heartbeat interval disabled by default. Without heartbeat.interval.ms, a connector watching a quiet table stops advancing its LSN in the WAL. PostgreSQL interprets this as an active consumer and retains WAL segments indefinitely. Disk grows without bound. The connector reports healthy — it's technically doing nothing wrong, because nothing is happening. But "nothing happening" and "quiet source table" are indistinguishable to a connector without heartbeats.
• Adding a table to a running connector. After modifying table.include.list to capture a new reference table, the connector started streaming changes from the current LSN. No snapshot of existing rows. The initial snapshot.mode: initial had already completed on startup — adding a table afterward doesn't re-trigger it. New changes appeared correctly; all history was silently missing.
• Date and decimal wire encoding. Debezium defaults to base64-encoded byte arrays for decimals and days-since-epoch integers for dates. Both are valid CDC output. Both fail schema validation immediately against any contract expecting JSON numbers and ISO 8601 strings.

None of these appear in Debezium’s quickstart. The real lesson wasn’t about reading documentation more carefully — it’s that every CDC default is an implicit decision interacting with your schema, your consumer, and your monitoring. Each one needs an acceptance criterion verified on the data plane.

Design your CDC schema for what Debezium actually produces

The original payment_line.v1 schema had nested objects: supplier: {name, tax_id} and billing_period: {start, end}. Debezium's ExtractNewRecordState produces flat rows from a relational table. Nested JSON from flat columns requires a custom reshaping stage — a new component purely to preserve a JSON shape.

The schema also declared additionalProperties: false while the connector was injecting CDC metadata (op, db, table, lsn). Every real message would have failed validation.

Both problems were resolved in one ADR: flatten the schema to match what CDC actually produces, strip all metadata from the payload. The contract test suite includes regression fixtures for the nested shape and the leaked-metadata shape — so a future config change that re-introduces either breaks CI before it reaches a running pipeline.

Write your architecture decisions down — especially the wrong ones

ADR-0001 established: Kafka for cost events via CDC, Kinesis for market prices. The rationale was sound. The document also stated: “Both are supported by PyFlink 2.x.”

False. FlinkKinesisConsumer depends on SourceFunction, which Flink 2.0 removed. No Kinesis connector exists on the new Source API. This surfaced at runtime, after three project phases had been built on the assumption.

A bridge service now reads Kinesis via boto3 and republishes to Kafka. A new ADR documents the correction. The value isn’t the bridge — it’s trivial. The value is that a written, falsifiable claim made the contradiction identifiable by reference instead of by debugging session.

The same pattern caught the partition key problem. ADR-0001 said both streams would use apartment_id. The market price schema has no apartment_id field — it's scoped by market segment. Caught at spec review, before any code was written, because the ADR existed as something checkable.

Build Flink state by hand before you trust an abstraction

This is the core of the project. The pricing engine needs a price per apartment per night. Costs are per apartment (no date). Market prices are per segment per night (no apartment). No shared key. No time alignment. Standard Flink joins — interval, temporal — don’t fit.

The solution is a KeyedProcessFunction keyed by market segment with two MapState leaves.

• Leaf 1: apartment_id → cost aggregate.
• Leaf 2: target_date → market snapshot.

A cost event updates Leaf 1 and fans out against every date in Leaf 2. A market event updates Leaf 2 and fans out against every apartment in Leaf 1.

This cross-product within a segment is the correct business semantics: a new invoice changes the calculation for all future nights of that apartment, and a market move for one night affects every apartment in its segment.

Built with KeyedProcessFunction instead of Table API — not because Table API can't express it, but because doing it by hand forces explicit reasoning about what Flink stores, when it emits, and what happens on replay. Table API generates retraction streams that the idempotent DynamoDB sink doesn't need.

Three risks were identified in writing before implementation and became actual tested code: stale-event overwrite protection (timestamp comparison before every MapState.put), eviction caps on both leaves (bounded state), and max_parallelism set explicitly to avoid freezing it at an arbitrary value on the first checkpoint — a default that silently limits all future rescaling.

Verify with chaos, not just design

A docker kill on the Flink TaskManager during active processing confirmed checkpoint restoration: cost_lines_count for a specific apartment continued growing after recovery without resetting or duplicating. The same test on the Iceberg CDC consumer — killed mid-stream, one record written while down, restarted — resumed from its checkpoint with exactly the expected rows. No duplicates. No gaps.

Single-writer is not optional

The schema says pricing decisions go to DynamoDB and Iceberg. Writing to both from the same Flink job is the dual-write problem from Kleppmann (DDIA, chapter 11): no atomicity between two independent sinks, no way to detect divergence after a partial failure.

Flink writes to DynamoDB only. Iceberg is populated via DynamoDB Streams CDC. Same pattern as upstream: the application writes to PostgreSQL, Debezium derives Kafka from the WAL. One writer per system, one derived copy. The pattern repeats because the problem repeats.

Say what’s missing — it’s a stronger signal than pretending everything works

Event-time and watermarks were never used. Processing-time was correct for this job (no windows to close), but it means zero practice with out-of-order handling or windowed aggregation. Of Flink’s three core primitives — State, Time, Windows — only State and Timers were covered.

No observability beyond logs and manual checks. Every incident was diagnosed with curl, kcat, and describe-table. The biggest gap against anything resembling production.

Never tested under real load. 100 apartments, 18 segments, synthetic volume. The hot-shard risk is documented and accepted, not observed.

A known bug remains unfixed: the cost-side fan-out doesn’t filter past nights. It doesn’t bite in practice, but it’s a latent race condition, deferred on purpose.

These are gaps, not secrets. A pipeline with 8 ADRs, 5 documented incidents, chaos-verified checkpoints, and an honest list of what’s still missing is a stronger signal than one that claims everything works perfectly.

Here’s the repository if you want to have a look and the steps and decisions made: https://github.com/moradabaz/pms-price-engine