Deduplication into upsert-kafka: retractions, tombstones and 3.1x message counts
orders_raw receives order events from an at-least-once producer, so about 5% of order_ids arrive twice with different updated_at values. The pipeline deduplicates with ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) filtered to rn = 1 and initially wrote to a kafka sink topic orders_dedup; deployment failed immediately with an error stating that the Kafka sink does not support consuming update and delete changes. The team switched the sink to upsert-kafka with PRIMARY KEY (order_id) in the DDL, and the job now runs. Two downstream consumers complain: a Python script counting messages in orders_dedup reports about 3.1 messages per distinct order, and some records have null values. orders_dedup was created with cleanup.policy=delete and retention.ms=604800000. No table.exec.state.ttl is set on the dedup job, which has been running for six weeks.
1. Explain the first deployment failure and exactly what changing the sink to `upsert-kafka` did to resolve it.
Read the worked answer
Non-windowed ROW_NUMBER() deduplication is a stateful operator on an updating stream: when a record with the same order_id but a later updated_at arrives, the previously emitted row is retracted (-U) and the new row is emitted (+U). The planner therefore produces a changelog, and a plain kafka sink is append-only - it can consume insert-only streams, so the job is rejected at plan time. upsert-kafka declares the table as a changelog: the DDL's primary key must be present, it is used as the Kafka message key, +U inserts and -U/+U pairs become upserts of the same key, and a -D becomes a tombstone with a null value. Because Kafka orders records within a partition, all versions of one order_id land in the same partition and arrive in order, so last-value-wins holds. The result is that the sink topic carries the full change history, not the final table - that distinction explains the next complaint.
2. Why does the Python consumer see 3.1 messages per order and occasional nulls, and what must change on the topic and in the consumer contract?
Read the worked answer
The consumer is reading the changelog rather than the state. Every duplicate produces additional messages, and every update produces a retraction, so raw counts exceed distinct orders by a large factor; null payloads are tombstones or delete records the script does not handle. The consumer contract must be upsert semantics: keep the last value per key, skip null values, and read with isolation.level=read_committed so uncommitted or aborted transactions are invisible. The topic itself is configured wrong for that contract: with cleanup.policy=delete and 7-day retention, intermediate versions and tombstones persist for a week, so any consumer starting mid-stream replays stale versions before reaching current ones. Set cleanup.policy=compact, optionally compact,delete with delete.retention.ms long enough for the slowest consumer, so the topic converges to one value per key. If downstream systems cannot be taught changelog semantics, materialize the deduplicated table through a connector with primary-key upserts instead of asking every consumer to interpret Kafka messages.
3. What is the state-growth risk of this dedup, and how would you choose a state TTL without silently breaking deduplication?
Read the worked answer
The dedup state is keyed by order_id and, without a TTL, retains one row per order ever seen, so it grows linearly with the order history and will eventually dominate checkpoint time and recovery. Set table.exec.state.ttl slightly above the maximum plausible gap between the first occurrence of an order and its duplicate. To choose the value, measure it rather than guess: compute the 95th percentile of duplicate.updated_at - first_seen over a representative window of orders_raw, and add a margin for out-of-order arrivals. The trade-off is explicit - a duplicate that arrives after the key has expired is no longer recognised, so it passes through as a new insert and produces a duplicate downstream. Shorter TTL means smaller state and weaker deduplication; there is no setting that gives both. Also remember cleanup is lazy, so state can exceed the configured horizon before compaction runs, and size memory for that margin rather than for the exact TTL bound.