PMS Price Engine
A dynamic pricing engine for vacation rentals built to learn streaming for real: PyFlink, Kafka, CDC with Debezium, keyed state, exactly-once checkpointing, and a pipeline designed spec-first with ADRs and live auditing.
What this project demonstrates
Select any card to read how it was solved and why.
Watermarks and event time
The setup
The Market Pulse job aggregates in 5-minute tumbling windows on event time, not processing time: bookings by created_at and market prices by collected_at. The watermark is bounded out-of-orderness of 30 seconds, because Kafka gives no ordering guarantee across different apartments or segments.
The minimum rule
An operator with several inputs takes the minimum watermark of all of them. It is deliberately conservative, since closing a window early would drop a late event that belonged to it. The price is that one input that stops advancing freezes every window of every key.
Two causes, two fixes
booking-events.v1 has a single Kafka partition while the job runs with parallelism 4, so three subtasks never receive a single event and pin the watermark from the first second. Wrapping that in with_idleness would hide it, so the fix is parallelism 1 on that source. A real quiet period is a different problem, and there with_idleness(60s) applies, which is 2x the longest gap the generator can produce (30s). The market source has 4 partitions and a deterministic 60s tick, so it gets with_idleness(120s) with no structural fix needed. Both thresholds are settings, because a real booking source will not behave like the mock.
Late data
Allowed lateness is 1 minute: a record inside it updates and re-emits the window. Anything later goes to a late-data side output that is logged, never silently dropped.
Rejected
Applying the same idleness to both sources "just in case" would have patched the symptom on one of them and guessed the number on both.
Exactly-once checkpointing
How it works
Flink injects a checkpoint barrier at the sources and each stateful operator snapshots its state as the barrier passes. When every operator has confirmed, the checkpoint is complete and becomes the recovery point. That is what gives EXACTLY_ONCE: operator state reflects every input record once, consistently across the whole pipeline.
RocksDB plus S3
RocksDB is the fast working copy on the TaskManager local disk, so state size is bounded by disk instead of RAM. S3 is the durable copy, and checkpoints are incremental, so only the SST files that changed are uploaded. The reads that RocksDB adds are cheap here: the workload is mostly writes, access is by exact key, hot blocks sit in the block cache and there is no network hop.
Tested by killing it
I killed a TaskManager with SIGKILL during a live run. The job restored from the last checkpoint in a few seconds, the checkpoint id kept climbing from 29 to 46, and the state stayed around 770 KB. At baseline the state was 770 KB but only 616 KB was uploaded, which is the incremental checkpoint at work.
What the test taught me
A job restart does not restart the TaskManager. Flink only recovers the job, and bringing the process back is the job of the infrastructure layer (Kubernetes, YARN or a restart policy). One TaskManager had been dead for about 37 minutes before I noticed, only because I was polling by hand. A checkpoint id that stops advancing is the early signal, which is what the Job health panel reads from /jobs/{id}/checkpoints, and a real deployment would alert on it.
Keyed state with a computed TTL
Why state at all
Debezium and Kafka can redeliver a record, for example after a restart from a checkpoint, and a duplicated booking would be summed twice inside a window. Writing the same result twice to DynamoDB is harmless because of the overwrite, but double-counting an input inside the aggregation is not, since the wrong total would be written perfectly idempotently.
The mechanism
A KeyedProcessFunction before the windows keeps a ValueState[bool] per booking_id. Unlike a window, keyed state has no automatic lifecycle: nothing cleans it up, so without a TTL every booking id ever seen would grow in RocksDB forever. Cancellations use their own independent state.
Where the TTL comes from
There are two replay cases. A same-day redelivery or job restart carries a recent timestamp, still inside the window, so it truly needs dedup. A full Debezium resnapshot carries rows that are weeks old, and the window discards them as late data before dedup state is ever consulted. So the TTL only has to cover the first case: window size (5 min) plus allowed lateness (1 min) = 6 minutes.
Rejected
Checking updated_at IS NULL fails exactly on the resnapshot it has to survive, because a row that was never updated looks like a new insert. A TTL of weeks pays for unbounded state to protect a case the window already handles for free. Relying on the sink overwrite does not stop a duplicate being summed upstream.
Async I/O enrichment
The goal
The average profit per booking needs the cost context that the pricing job already wrote to the price_decision table in DynamoDB. Each confirmed booking triggers a point lookup by apartment_id and check-in date, and the profit is revenue * (1 - p) - total cost, using the fields stored in that decision.
Why not a third CDC bridge
The usual pattern in the project was a keyed join fed by another bridge service and topic. For a value with a stable key in a store that is already queryable, that would be a third service and a third topic for one lookup. Async I/O is also a Flink capability the pricing job never used, and this phase was designed to learn windows and Async I/O.
The trade-off, accepted on purpose
Every booking now pays real I/O latency and depends on DynamoDB being available for the enrichment step. A keyed-state join would have no per-event I/O once warm. For a 5-minute analytics metric, one service fewer was worth more than the lookup latency, and AsyncDataStream timeout and retry cover the failures. unordered_wait is used because the window does not care about arrival order.
Missing data is not zero
If no price_decision exists yet (a brand-new apartment or a date past the pricing horizon) the booking still counts in the booking count but is left out of the profit average, logged as a warning. A missing cost basis and a genuinely zero profit are different things, and mixing them would corrupt the average. The window also reports how many bookings were excluded.
Spec-driven development
Three kinds of document
There are 26 phase specs written before the implementation, with acceptance criteria, non-goals and explicit design decisions. There are 18 ADRs, each one a single architectural decision with the alternatives that were rejected. And there is an audit diary, which records what was actually verified against the live stack and when. The spec says what was going to be built and the diary says what was proven.
Verified live, not assumed
Acceptance criteria were run against the running stack, and that found real errors in the specs themselves. For example, the command drafted to inspect the replication slot failed because pgoutput emits binary output, so the spec had to switch to pg_logical_slot_get_binary_changes.
The lesson that repeated three times
A green health check does not mean the data flows. The Debezium connector reported RUNNING while publishing zero messages because the topic name had silently diverged from the spec. Another time a single unresolvable heartbeat topic stalled the shared producer for every topic. A third time DynamoDB reported a stream ARN that no longer existed. Every control-plane check now has a data-plane check next to it, such as replication slot lag or consuming the topic, and each incident is written up with the situations where it can happen again.
Cost discipline in architecture
Storage and state in real AWS, compute local
ADR-0010 puts S3, Kinesis, DynamoDB and the Glue catalog on real AWS, because that is where LocalStack differs from reality (IAM enforcement, Glue tier limits, DynamoDB Streams ARNs). Kafka, Debezium, Flink and Postgres stay in Docker Compose and talk to AWS by configuration alone.
A one-shot deployment
terraform apply, run the demo, then terraform destroy straight away, confirmed with an empty state list. AWS Budget alerts at $5 and $10 must exist before the first apply, as a hard precondition.
Rejected
Managed Kafka or managed Flink bill per node-hour even at the smallest tier and would burn a $5 to $10 budget in minutes. An always-on deployment has no uptime requirement to justify itself, and staying on LocalStack forever would leave the AWS learning goal untested.
The same reasoning in the other choices
Kafka carries the Debezium payment events because it is Debezium native target, while market prices go straight to Kinesis, and a small bridge republishes them to Kafka because Flink 2.x has no Kinesis source. PyFlink was chosen over Java because the stack is Python, the learning goal is stream processing and not the JVM, and the volume is around 100 apartments. The ADR itself says to revisit it at higher throughput.
Architecture
CDC from Postgres via Debezium, two independent PyFlink jobs on the same cluster (pricing and Market Pulse), DynamoDB as the hot path, and Apache Iceberg as the cold path, transformed with dbt.
