Duplicate Records in Tableflow

🧠 Free Recall ❓ Quiz 🔍 Cues 🔮 Simulation 🥷 ShadowBox 📐 Encode Cards

How Tableflow works
Tableflow is a Confluent Cloud service that materializes a Kafka topic as an open table (Iceberg or Delta) without you running a sink connector or ETL job. It handles the work you'd otherwise do by hand: cleaning, schema management, materializing CDC streams, transformation, compaction, and writing Parquet/Iceberg files. Mechanically: Confluent
You enable it per topic. Tableflow reads the topic and its Schema Registry schema, and writes Parquet data files plus Iceberg metadata/manifests into either Confluent-managed storage or your own bucket.
Tables are exposed through Confluent's embedded Iceberg REST catalog, or synced to AWS Glue, Snowflake Open Catalog, or Databricks Unity Catalog, so Spark/Trino/Athena/Snowflake/DuckDB query them as normal tables. ConfluentConfluent
It also manages schema evolution, file compaction, and catalog registration on an ongoing basis. Snowflake

There are two write modes. Append treats every Kafka record as a new row — good for event streams. Upsert is the one you're asking about: the Kafka message key and partition number form the composite primary key of the table, and Tableflow retains the most recent version of data for each key. This is the same behavior as a Flink table in upsert mode. Tableflow maintains an additional index table to make this perform at scale, and if your key exceeds 256 bytes it uses a hash of the key for the uniqueness check. Updates and deletes are written as separate delete files (merge-on-read), which Tableflow merges and compacts automatically; compatible engines reconcile the upserts at query time. Deletes require tombstones — a record with a key and an empty value. Confluent + 2
One consequence worth internalizing: Tableflow doesn't own a re-partitioning step. It dedupes within what Kafka already gave it, using Kafka's key bytes and Kafka's partition assignment as ground truth.

Why duplicates still show up
1. The primary key includes the partition number. This is the single biggest source. If the same logical entity lands in two different partitions, you get two rows, both "correct" from Tableflow's point of view. Confluent calls this out explicitly for partition-count increases: raising the partition count on a compacted topic used with upsert mode can produce duplicate rows, because producers may route keys that previously went to one partition to a different one. Other paths to the same outcome: Confluent
Different producers using different partitioners (Java client default vs. librdkafka vs. a custom one), or a client that assigns partitions explicitly.
A backfill or migration job writing the same keys with a different distribution than the live producer.
Cluster linking / Replicator / a Flink job that repartitions with a different bucketing scheme.
2. The key bytes aren't stable. Tableflow compares serialized keys. "1001" as a string key and 1001 as an Avro int are different keys. So are keys that differ in whitespace, case, field order in a composite string, or that switched serializers mid-life. Also note the gotcha that if the key fields exist in both the Kafka key and the value, the values from the record value are used, not the key — so key/value drift can produce surprising row identity. Confluent
3. The Kafka key isn't the business key. If the producer keys on event_id, order_id + version, or a per-attempt UUID, then a retried or re-emitted logical record arrives under a new key. Upsert mode dedupes perfectly and you still get two rows, because they genuinely are two keys.
4. CDC configuration gaps. With Debezium sources in upsert mode you should have tombstones.on.delete set to true, or deleted source rows are never removed from the table. Debezium snapshot-plus-streaming overlap, or an un-unwrapped Debezium envelope, can also change effective key semantics versus what you expect. And if the topic is actually in append mode, every CDC event is a row by design. Confluent
5. Reader-side false duplicates. Because upsert tables are merge-on-read, an engine that doesn't apply Iceberg v2 equality delete files (or that reads the Parquet files directly, bypassing Iceberg metadata) will show both the old and new versions. Same if it's pinned to a stale snapshot. Before checking your pipeline, confirm the engine reconciles deletes.
6. Ordering, not duplication. Worth separating out: with multiple producers or out-of-order arrival, "most recent" means last-written, not latest event time. That yields a wrong surviving version rather than an extra row — a different bug with a similar smell.
A quick diagnostic to tell case 1 apart from cases 2/3: group by your business key and count distinct Kafka partition values. If duplicates span multiple partitions, it's partition drift. If they sit in one partition with different key bytes, it's a keying problem.

Deduplication options
Prevent it at the producer. Cheapest by far. Key on the true business primary key with one canonical serialization, standardize the partitioner across every producer, enable idempotent producers, never assign partitions manually, and treat partition-count changes on upsert-backed topics as a breaking change. If duplicates already exist from a partition increase, Confluent's own remedy is to send tombstones for the affected keys to their original partitions — both log compaction and Tableflow then drop the stale rows, leaving only the record from the new partition. Confluent
Normalize in Kafka before Tableflow. The most robust structural fix: run a Flink SQL statement that re-keys onto the business key into a new topic with changelog.mode = 'upsert', then enable Tableflow upsert on that topic. The primary key becomes an explicit contract, and key→partition mapping is consistent because Flink hash-distributes on the primary key. If you also need to collapse genuine duplicates with different keys, the standard Flink pattern is ROW_NUMBER() OVER (PARTITION BY business_key ORDER BY ts DESC) filtered to rank 1 — budget for state size and set a TTL. This costs a compute pool and a second topic, but it moves dedup upstream of everything, not just Iceberg.
Deduplicate at read time with a view. Non-destructive, zero pipeline change, and it works today: define a view over the Tableflow table with ROW_NUMBER() OVER (PARTITION BY business_key ORDER BY <tiebreaker> DESC) = 1, and point consumers at the view. For the tiebreaker, prefer a source-system ordering column (CDC LSN/SCN, updated_at) over Kafka timestamp, and use Kafka offset only as a last resort since it isn't comparable across partitions. Downside: the dedup cost is paid on every query.
Materialize a curated downstream table. Treat the Tableflow table as bronze and maintain your own silver table keyed on the business key via MERGE INTO (Spark, Trino, Athena engine v3, Snowflake) or dbt incremental with a merge strategy, reading incrementally from Iceberg snapshots. This gives you copy-on-write files that every engine reads cleanly, plus a place to enforce ordering rules. Important caveat: Tableflow-managed tables are owned by Tableflow — don't run DML against them; write to a separate table you control.
Monitor for it. Regardless of which fix you pick, a scheduled check of count(*) vs. count(distinct business_key), plus a check for business keys appearing under more than one partition, will catch regressions when someone scales partitions or ships a new producer.
If you tell me which situation you're in — CDC vs. application-produced, and which engine is querying — I can narrow this to a specific recommendation and sketch the actual SQL.

Situation A: CDC-produced topics (Debezium / Oracle XStream / etc.)
Where duplicates come from
Delete events that aren't tombstones. This is the most common CDC-specific one. A Debezium delete arrives as a record with a real value (op='d', after=null), not an empty value. Tableflow only removes a row on a true tombstone, so the row survives as a ghost — and if the source later reinserts that PK, you now have a stale row plus a live row that look like duplicates. This is why the docs say that with Tableflow upsert mode you should set tombstones.on.delete to true, otherwise rows won't be deleted from the table when they're deleted from the source (and conversely, for append mode you want it false or you'll get null-value rows landing in the table). Confluent
Soft deletes. If your source marks is_deleted = true rather than deleting, no tombstone is ever produced. That's not a bug, but it means "current state" queries must filter, and analysts read it as duplication.
The envelope. If you don't unwrap, the table has before/after/op/source nested structs. Row identity still works (key is unchanged), but every dedup query has to reach into after.*, and a WHERE on business columns silently misses rows. The documented options are either to decode with Flink — copy the source table DDL, set changelog.mode to upsert and value.format to avro-registry, then INSERT INTO <new-topic> SELECT * FROM <source-topic> — or to set after.state.only to true on the Debezium connector so it emits messages resembling the source table schema directly. Confluent
Key definition changes. Debezium keys on the source table's primary key. Change message.key.columns, add a column to a composite PK, switch the key converter (Avro → String), or re-snapshot a table after a PK change, and the same logical row arrives under different key bytes. Old rows are orphaned, never overwritten.
Partition scaling. CDC topics usually start at 1–6 partitions and get scaled when throughput grows. This is exactly the documented duplicate-row scenario, and the documented remedy is to send tombstones for the affected keys to their original partitions, after which both log compaction and Tableflow drop the outdated records, leaving only the record from the new partition. Confluent
Multi-table routing. Outbox pattern or EventRouter funnelling several tables into one topic risks key collisions across entity types unless you namespace the key.
Prevention checklist
tombstones.on.delete = true
after.state.only = true, or unwrap with Flink upsert mode before Tableflow
Freeze partition count on any CDC topic backing an upsert table
Treat PK/key-converter changes as a new topic + re-materialize, not an in-place edit
Snapshot before enabling Tableflow, so the initial snapshot collapses in Kafka rather than in Iceberg
Read-time dedup for CDC
The important part is the tiebreaker. Use source-database ordering, not Kafka ordering:

sql
CREATE OR REPLACE VIEW customers_current AS
SELECT * FROM (
  SELECT c.*,
         ROW_NUMBER() OVER (
           PARTITION BY customer_id
           ORDER BY source_lsn DESC,      -- Postgres LSN / Oracle SCN / MySQL binlog pos
                    source_ts_ms DESC     -- fallback
         ) AS rn
  FROM tableflow.cdc.customers c
) WHERE rn = 1
  AND op <> 'd';                          -- if envelope retained
If you kept the envelope, project after.* and carry source.lsn / source.ts_ms up as top-level columns in the view so downstream consumers never see the envelope. Run DESCRIBE on the table first — the exact naming of the Kafka key and record-metadata fields Tableflow adds varies by mode and format, and you'll want the partition column available for diagnostics.
Ordering caveat: source.ts_ms has millisecond granularity, so two updates inside the same millisecond tie. LSN/SCN is monotonic and is the right choice when available.

Situation B: Application-produced topics
Where duplicates come from
The key isn't the entity. Applications very often key on event_id, a UUID, or entity_id + version. Upsert mode then dedupes perfectly and you still get one row per event, because every event genuinely is a distinct key. This is the single most common cause here, and it's a modelling problem, not a Tableflow problem.
Retries that regenerate the key. Mobile client or API gateway retries where the UUID is minted per attempt rather than once at the edge. Two records, two keys, identical payload. No key-based mechanism can catch this — you need a deterministic idempotency key generated once upstream, or content-hash dedup.
Producer retries without idempotence. enable.idempotence = false plus a retry produces the same key twice in the same partition. Notably this is the benign case: upsert collapses it. Worth knowing so you don't chase it.
Heterogeneous producers. A Java service and a Python service writing the same topic hash keys differently in some configurations; a service that calls send(topic, partition, key, value) explicitly bypasses hashing entirely. Same key, two partitions, two rows.
Mutable natural keys. Keying on email, username, or slug. The user changes it, the app produces under the new key, and the old row sits there forever as a phantom duplicate. Fix requires a tombstone to the old key — which means the app has to remember it.
Migrations and replays. Dual-write during a service cutover, or replaying an archive topic into the live topic with different keying. Also relevant: Flink's upsert-kafka sink is at-least-once by default with checkpointing, so it may write duplicate records with the same key — but since the last record per key wins on read, this is idempotent in practice. Benign again. Apache
No key at all. If records are unkeyed, upsert mode has nothing to identify rows with. Don't rely on inferred behaviour here — either set a key or use append mode.
Prevention checklist
Key on the stable entity identifier (order_id, device_id), never on the event
Generate any idempotency key once, at the system edge, and propagate it
enable.idempotence = true and a single agreed partitioner across all producer languages
Never pass an explicit partition
Freeze partition count; if you must change it, re-key into a new topic via Flink and re-materialize
Emit a tombstone to the old key whenever a natural key changes
Read-time dedup for app-produced data
Two different queries depending on what you want:
Latest state per entity (when the topic is a change stream keyed on the event):

sql
SELECT * FROM (
  SELECT o.*,
         ROW_NUMBER() OVER (
           PARTITION BY order_id
           ORDER BY updated_at DESC, event_id DESC   -- event_id breaks ties deterministically
         ) AS rn
  FROM tableflow.app.orders o
) WHERE rn = 1;
Collapsing true retry duplicates (different keys, same payload) — dedup on business content rather than key:

sql
SELECT * FROM (
  SELECT p.*,
         ROW_NUMBER() OVER (
           PARTITION BY idempotency_key                 -- preferred
           -- or: md5(concat_ws('|', account_id, amount, cast(occurred_at as varchar)))
           ORDER BY ingest_ts ASC                       -- keep the first occurrence
         ) AS rn
  FROM tableflow.app.payments p
) WHERE rn = 1;
Note the flip: for state you keep the last row; for retry-dedup on immutable events you keep the first. Getting that backwards is a real and easy bug.

The structural difference between the two
For CDC, the entity key is given to you by the source database and is trustworthy — your problems are almost entirely delete semantics and configuration. For application-produced data, delete semantics rarely matter but the key itself is a design choice your producers got to make, so your problems are almost entirely key modelling.
That leads to different default recommendations. CDC: fix the connector config and unwrap upstream, then a thin view is usually enough. Application-produced: re-key in Flink into a purpose-built upsert topic, because you're not going to get every producer team to agree, and the re-key also normalises partitioning for free.
Diagnostic worth running once on either

sql
-- Which failure mode am I in?
SELECT entity_key,
       COUNT(*)                        AS rows_,
       COUNT(DISTINCT kafka_partition) AS partitions_,
       COUNT(DISTINCT kafka_key)       AS distinct_keys
FROM tableflow.<db>.<table>
GROUP BY entity_key
HAVING COUNT(*) > 1
ORDER BY rows_ DESC
LIMIT 50;
partitions_ > 1 means partition drift. distinct_keys > 1 within one partition means a keying or serialization problem. Both equal to 1 with rows_ > 1 means your reader isn't applying the equality delete files — check the engine, not the pipeline.
──────────────────────
Q: What are techniques to send tombstone records at scale in the case of 500M rows?

Sending Tombstone Records at Scale (500M Rows)
This is a genuinely hard operational problem. Sending 500M tombstones isn't just a "write a script" task — you have to think about Kafka broker load, log compaction behavior, Tableflow's merge/compaction pipeline, and whether you can afford to do this incrementally or need it done in a bounded window. Let me walk through the full landscape.

First: Clarify What You Actually Need to Tombstone
Before producing 500M records, ask whether you need all of them:
Are all 500M rows genuinely deleted at the source? If this is a sync/repair operation (e.g., after a partition count increase), you may only need tombstones for the orphaned partition-drift duplicates, not every row.
Is this a full table wipe? If so, deleting the topic and re-materializing from scratch is almost always better than sending 500M tombstones.
Can you scope it by time range or key range? Even reducing the problem to 50M targeted tombstones is a 10x improvement.
If you genuinely need all 500M, read on.

Core Mechanics to Understand First
A tombstone is just a Kafka produce call with:
The exact same key bytes as the record you want to delete (serialization must match)
A null value (not an empty string, not {} — literally null)
Ideally sent to the same partition the original record lives in
Tableflow's upsert mode uses (key bytes, partition) as the composite primary key. So if you're cleaning up partition-drift duplicates, you need to send the tombstone to the old partition explicitly — not let the partitioner route it naturally. This is critical and easy to get wrong at scale.

Technique 1: Kafka Streams or Flink — Streaming Tombstone Generator
Best for: targeted deletes where you have a source of truth for "what should be deleted"
Write a Flink or Kafka Streams job that:
Reads from a "deletion manifest" topic or table (your list of keys to delete)
Produces tombstone records to the target topic with explicit partition assignment
-- Flink: read a deletion manifest and produce tombstones
INSERT INTO target_topic_upsert
SELECT 
  business_key,
  CAST(NULL AS ROW<...>) AS value  -- null value = tombstone
FROM deletion_manifest
WHERE needs_tombstone = true;
Advantages:
Naturally rate-limited by Flink's parallelism settings
Checkpointed, so restartable — critical for 500M records
Can be targeted precisely
Flink's upsert-kafka sink handles the null value correctly
Gotcha: Make sure your Flink job is configured to send to the correct partition when dealing with partition-drift cleanup. You may need a custom partitioner or explicit partition routing.

Technique 2: Kafka Producer Application with Explicit Partition Routing
Best for: partition-drift cleanup where you know the old partition assignments
This is the scenario the document mentions: after a partition count increase, you need to send tombstones to the original partitions. You can't let the partitioner route freely because it'll send to the new partition (which already has the correct record).
from confluent_kafka import Producer
import concurrent.futures

producer = Producer({
    'bootstrap.servers': '...',
    'enable.idempotence': True,          # critical for reliability
    'acks': 'all',
    'compression.type': 'lz4',
    'batch.size': 1048576,               # 1MB batches
    'linger.ms': 50,                     # allow batching
    'queue.buffering.max.messages': 500000,
})

def send_tombstone(key_bytes, partition):
    producer.produce(
        topic='your-topic',
        key=key_bytes,
        value=None,              # true null = tombstone
        partition=partition,     # explicit routing for drift cleanup
    )

# Process in parallel batches
with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:
    for batch in read_keys_in_batches(batch_size=10_000):
        executor.submit(process_batch, batch)
Throughput considerations:
A well-tuned producer can do ~500K-1M small messages/sec
500M tombstones at 500K/sec = ~17 minutes of pure produce time
But broker-side compaction and Tableflow processing lag behind — see below
Technique 3: Spark Job Reading from Iceberg Snapshot
Best for: when your source of truth for "what to delete" is the Tableflow Iceberg table itself
If you need to tombstone everything currently in the table (e.g., a full re-sync):
from pyspark.sql import SparkSession
from confluent_kafka import Producer, KafkaException
import json

spark = SparkSession.builder.getOrCreate()

# Read keys directly from the Iceberg table
# This gives you the exact key bytes and partitions Tableflow knows about
keys_df = spark.read.format("iceberg") \
    .load("your_catalog.db.table") \
    .select("_kafka_key", "_kafka_partition") \
    .distinct()

# Repartition for parallelism
keys_df.repartition(200).foreachPartition(send_tombstones_for_partition)
The key insight here: reading from the Iceberg table gives you the exact key bytes and partition assignments that Tableflow used — no guessing about serialization format.
Important: Don't run DML against Tableflow-managed tables (the document warns about this). You're only reading here, which is fine.

Technique 4: Compacted Topic Reset (Nuclear Option)
Best for: full wipe where you want to start fresh
If you want to delete all 500M rows:
Delete and recreate the topic (if you control the pipeline end-to-end)
Re-enable Tableflow on the new topic
Re-produce current state from your source of truth
This avoids sending 500M tombstones entirely. Log compaction still has to process those records; Tableflow still has to process them. A clean topic is dramatically cheaper.
Alternatively, if you can't delete the topic:
Produce all current "live" state as fresh records (not tombstones) to a new topic
Enable Tableflow upsert on the new topic
Deprecate the old topic
Technique 5: Tiered / Incremental Approach
Best for: when you can't afford a big-bang operation
Rather than sending 500M tombstones in one shot:
Identify and tombstone the highest-impact duplicates first — run the diagnostic query from the document, find keys appearing in multiple partitions, tombstone those specifically
Batch by key range or time range — e.g., tombstone keys A-M this week, N-Z next week
Use a Flink job with a rate limiter to avoid overwhelming the broker
-- Flink: rate-controlled tombstone emission
-- Process in windows to avoid broker saturation
INSERT INTO target_topic
SELECT key, CAST(NULL AS ...) as value
FROM deletion_manifest
WHERE key_range BETWEEN :start AND :end
  AND batch_id = :current_batch;
Critical Operational Concerns at This Scale
1. Broker-Side Log Compaction Lag
Sending 500M tombstones doesn't immediately free space or remove rows from Tableflow's perspective. Log compaction runs asynchronously. Key settings to tune:
log.cleaner.min.compaction.lag.ms — how long before a segment is eligible for compaction
log.cleaner.threads — increase this temporarily
log.cleaner.io.max.bytes.per.second — throttle to avoid I/O saturation
Tombstones themselves are only removed after compaction runs twice (once to compact them in, once to clean the tombstone itself after delete.retention.ms).

2. Tableflow Compaction Lag
Tableflow will see the tombstones as delete files (merge-on-read). It processes and compacts these asynchronously. At 500M
──────────────────────
Q: Why is key "bytes" specifically used here. What does this refer to?

Why "Key Bytes" Specifically
The term "key bytes" is precise and deliberate — it's pointing at something that trips people up constantly.

Kafka Keys Are Raw Binary
At the Kafka protocol level, a message key is not a string, not an integer, not an Avro record — it is just a sequence of bytes. Kafka itself is completely agnostic about what those bytes mean. It doesn't know if they represent a customer ID, a UUID, or a JSON object. It just stores and compares them as opaque binary data.
When Tableflow needs to determine "is this message an update to an existing row, or a new row?", it can only do what Kafka does: compare the raw byte sequences. It has no semantic understanding of what the key means — only whether the bytes match.

Why This Matters Enormously
This is where the subtlety lies. Two keys that are logically identical to a human can be bytewise different to Kafka and Tableflow:
| Scenario | Key 1 bytes | Key 2 bytes | Same? | |---|---|---|---| | Integer 1001 as Avro vs. string "1001" | 0x00 0x00 ... (Avro encoded) | 31 30 30 31 (ASCII) | ❌ No | | "ACME Corp" vs "acme corp" | different casing | different casing | ❌ No | | JSON {"id":1,"type":"order"} vs {"type":"order","id":1} | field order differs | field order differs | ❌ No | | Avro schema v1 encoding vs v2 encoding | schema ID bytes differ | schema ID bytes differ | ❌ No |
In every case above, a human (or your database) would say "same entity." Tableflow says "different key → different row" and you get a duplicate.

The Serializer Is the Key
This is why the document emphasizes canonical serialization as a prevention strategy. The serializer your producer uses is what determines the byte sequence. If:
One producer uses the Java Kafka client with Avro serialization
Another uses librdkafka with string serialization
A third uses a custom JSON serializer
...they can all be writing what they think is customer_id = 1001, but they're writing three different byte sequences, producing three separate rows in Tableflow's upsert table.

The Partitioner Connection
This is also why key bytes are mentioned in the context of partitioning. Kafka's default partitioner uses a hash of the key bytes to decide which partition a message goes to. If the byte representation of a key changes (e.g., you switch serializers), the hash changes, the partition changes, and now Tableflow sees what it considers a new key in a new partition — a double duplicate problem.

The 256-byte Hash Note
The document also mentions that if your key exceeds 256 bytes, Tableflow uses a hash of the key bytes for the uniqueness check. This is another place where "bytes" matters — it's the length in bytes of the serialized key, not the logical length of your key string. An Avro-encoded composite key with several fields can easily exceed 256 bytes even if the logical values are short.

In Short
"Key bytes" is used rather than just "key" to remind you that Tableflow operates one abstraction layer below your mental model. You think in terms of customer IDs and order numbers; Tableflow thinks in terms of byte arrays. The gap between those two levels of abstraction is exactly where duplicate rows are born.
──────────────────────
Q: How should one intuit this amongst the different types of guarantees when using Kafka (exactly once, at least once, etc.) on why this happens

Why Partition-Count Changes Cause Duplicates Despite Kafka's Delivery Guarantees
This is a genuinely important conceptual distinction, and the confusion is understandable because people often conflate delivery guarantees with identity guarantees. They're different axes entirely.

The Delivery Guarantee Axis (What Kafka Actually Promises)
Kafka's exactly-once semantics (EOS), at-least-once, and at-most-once are all about whether a message gets written to the log, and how many times. Specifically:
At-most-once: A message might be lost, never duplicated.
At-least-once: A message will definitely land, but might be written more than once (due to producer retries).
Exactly-once: With idempotent producers + transactional APIs, a message is written exactly once to a given partition, even across retries.
The critical thing to internalize: all of these guarantees are scoped to a single partition. Exactly-once means "this message appears exactly once in partition 3." It says absolutely nothing about which partition the message goes to, or whether the same logical entity has ever appeared in a different partition before.

The Identity Guarantee Axis (What Tableflow Needs)
Tableflow's upsert mode needs something different: it needs to answer the question "have I seen this entity before?" Its answer is: "yes, if I've seen the same (key bytes, partition number) combination before."
The composite primary key being (key_bytes, partition) is the crux of everything. So when you increase partition count:
A producer using the default partitioner computes partition = hash(key) % partition_count
Before the increase: hash("order-123") % 6 = partition 2
After the increase: hash("order-123") % 12 = partition 7
Tableflow sees ("order-123", partition 2) and ("order-123", partition 7) as two distinct primary keys
Exactly-once semantics did their job perfectly — the message was written exactly once to partition 7. But Tableflow has no way to know that partition 7's record is the "same entity" as partition 2's record. From its perspective, they're genuinely different rows.

The Intuition: Kafka Is a Log, Not a Database
Here's the mental model shift that makes this click:
Kafka's guarantees are about the integrity of the log as a log. They ensure durable, ordered, non-duplicated writes within a stream. Kafka does not have a concept of "this message supersedes a previous message about the same entity." That's not what it's for.
Exactly-once in Kafka is analogous to a transaction in a write-ahead log — it guarantees the write happened cleanly. But it doesn't give you the semantics of a MERGE INTO or UPSERT statement in a database, which requires a stable identity lookup across all prior writes.
Tableflow is trying to synthesize database upsert semantics on top of a log. To do that, it needs a stable, consistent identity signal. It uses (key_bytes, partition) as that signal. The moment partition assignment becomes inconsistent across time — which partition scaling directly causes — that identity signal breaks, and no delivery guarantee in Kafka can fix it, because delivery guarantees were never about identity.

Why Log Compaction Doesn't Save You Either
You might think: "But I'm using a compacted topic — doesn't Kafka's log compaction deduplicate by key?"
Yes, but only within a partition. Kafka's compaction runs independently per partition. It will happily retain one record for "order-123" in partition 2 and one record for "order-123" in partition 7, because from Kafka's perspective those are records in two different partitions — compaction has no cross-partition identity concept either.
This is exactly why the document notes the problem is specifically acute for compacted topics with upsert mode: you have two systems (Kafka compaction and Tableflow upsert) that both deduplicate, but both do so within a partition boundary, so cross-partition duplicates fall through both nets.

The Hierarchy of What Each Guarantee Covers
| Concern | Covered by | |---|---| | "Did the message make it to the broker?" | Delivery guarantee (EOS/ALO/AMO) | | "Is this message written exactly once to its partition?" | Idempotent producer + EOS | | "Will this message always go to the same partition?" | Nothing in Kafka — this is a producer-side contract | | "Is this the latest version of this entity?" | Tableflow upsert — but only within a stable (key, partition) identity | | "Is this the same entity as a record in another partition?" | Nothing in Kafka or Tableflow — requires external re-keying |
The gap is in row 3. Kafka's guarantees are silent on partition stability across the lifetime of a topic. That's a producer-side contract you have to enforce yourself — which is why the document's advice is to treat partition-count changes on upsert-backed topics as a breaking change, not a routine operational action.

The Deeper Principle
Exactly-once semantics solve a reliability problem (will the write happen cleanly?). The duplicate-row problem here is an identity problem (what makes two records refer to the same entity?). These are orthogonal. You can have perfect exactly-once delivery and still have unbounded identity duplicates, because EOS never claimed to know anything about entity identity. The fix lives upstream of Kafka's delivery layer — in how producers assign keys and partitions — or downstream in a Flink re-keying step that establishes a stable, partition-independent identity before Tableflow ever sees the data.
──────────────────────
Q: These are three themes on why the same key can exist on different partitions. How common are these themes in the wild and are there ways to mitigate them? In what use cases may this incidence (same key existing on multiple partitions) be harmful?

How Common Are These Themes in Practice?
Theme 1: Heterogeneous Producers with Different Partitioners
Very common. This is probably the most frequently encountered issue in organizations that have grown beyond a single team or a single language runtime. Here's why it bites so often:
The Java client's default partitioner changed behavior in Kafka 2.4 (from pure murmur2 hash to the "sticky partitioner") and changed again in 3.3 (back toward round-robin for unkeyed messages). So even two Java producers on different versions of the client can route the same key differently.
librdkafka (used by Python's confluent-kafka, Go, .NET, etc.) uses murmur2 but historically had subtle differences in how it handled null bytes in keys and composite key serialization.
Explicit partition assignment (send(topic, partition, key, value)) is shockingly common in legacy code where a developer "optimized" something years ago and nobody remembers why. The partition is hardcoded, and the key becomes decorative.
Microservice sprawl makes this worse: the orders service (Java), the reconciliation job (Python), and the data backfill script (Go) may all write to the same topic with no coordination on partitioning strategy.
Mitigation:
Establish a canonical partitioner contract at the topic level and document it as part of the topic's schema contract (alongside the Schema Registry schema). Treat partitioner choice as infrastructure, not a per-team decision.
Use a Kafka proxy or producer SDK wrapper that enforces the partitioner — internal platform teams often build this. All producers call the wrapper, not the raw client.
If you're using Confluent Schema Registry, you can enforce serialization format, but there's no equivalent enforcement for partitioning — that discipline has to be organizational.
For upsert-backed Tableflow topics specifically, consider routing all writes through a single authoritative Flink job that re-keys and re-partitions, so no producer ever writes directly to the Tableflow-backing topic. This is the "normalize in Kafka before Tableflow" pattern — it converts a social/organizational problem into a technical boundary.
Theme 2: Backfill and Migration Jobs
Extremely common, and disproportionately dangerous. Almost every data team eventually needs to:
Replay historical data after a bug fix
Backfill a new field by re-emitting old records
Migrate from one system to another with a dual-write period
Re-seed a topic after a retention-related data loss
The problem is that these jobs are usually written quickly, under time pressure, by someone who isn't the original author of the live producer. They often:
Use a different language/client than the live producer
Assign partitions explicitly for "efficiency" (parallelizing by partition)
Use a different key serialization (the live producer uses Avro, the backfill script uses a plain string)
Run concurrently with the live producer, creating a window where the same logical key exists in two partitions simultaneously
This is particularly insidious because the problem may not be visible immediately. The backfill finishes, everything looks fine, and then a query three weeks later returns duplicates that nobody can explain.
Mitigation:
Treat backfill jobs as first-class producer code, not scripts. They should use the same SDK wrapper, the same serializer, the same partitioner.
Before running a large backfill, tombstone the existing records first (for upsert topics), then re-emit. This ensures the old partition assignment is cleared before the new one lands.
Run backfills into a staging topic first, verify the partition distribution matches the live topic's distribution, then promote.
For Tableflow upsert topics specifically, consider pausing Tableflow during the backfill window or running the backfill into a separate topic and merging downstream — the document's "bronze/silver" pattern is exactly right here.
Theme 3: Cluster Linking / Replicator / Flink Repartitioning
Common in larger, more sophisticated architectures, but often underestimated. The specific failure modes:
Cluster Linking preserves partition assignment by design, so it's actually the safest of the replication tools for this problem — but it only works if the source and destination topic have the same partition count. If they differ, Cluster Linking can't be used and you fall back to MirrorMaker or Replicator, which rehash keys.
MirrorMaker 2 / Replicator rehash keys using the destination topic's partition count. If source has 6 partitions and destination has 12, keys will land in different partitions than they did at the source. This is well-documented but frequently forgotten.
Flink jobs are the sneakiest. A Flink INSERT INTO kafkasink SELECT * FROM kafkasource will repartition based on Flink's internal key distribution, which uses a different hashing scheme than Kafka's murmur2. Even if you set the same parallelism as the partition count, the mapping isn't guaranteed to match. The document's recommendation to use changelog.mode = 'upsert' on a Flink-produced topic is correct precisely because then Flink owns the key→partition mapping consistently.
Mitigation:
When replicating topics that back upsert Tableflow tables, use Cluster Linking with matching partition counts where possible.
If you must use MirrorMaker/Replicator, treat the destination topic as a new topic with a new lineage — re-materialize the Tableflow table from scratch rather than assuming continuity.
For Flink pipelines, always use upsert-kafka as the sink connector format (not kafka with manual key setting), and let Flink's upsert semantics own the deduplication contract.
When Is Same-Key-on-Multiple-Partitions Actually Harmful?
Not every use case cares equally. Here's a breakdown:

High harm:
1. Upsert-mode Tableflow tables (the core case here) This is the document's central concern. The composite primary key includes the partition number, so two partitions = two rows = data quality corruption in any downstream system that treats the table as a source of truth.
2. CDC pipelines feeding operational systems If a customer record exists in two partitions, a downstream MERGE INTO a data warehouse will either fail (PK violation) or silently pick one version, potentially the stale one. For regulated industries (finance, healthcare), this is a compliance issue, not just a data quality issue.
3. Aggregation and analytics on mutable entities If you're computing SUM(revenue) GROUP BY customerid and customerid 1001 has two rows from two partitions, your revenue figure is doubled for that customer. These bugs are particularly hard to catch because the error is proportional to the duplicate rate, which may be small enough to evade casual inspection.
4. Event-driven downstream consumers doing their own state management If a downstream Flink job or KStream app consumes the topic and maintains its own state keyed on the business key, it will see two separate state entries for the same entity, leading to split-brain state.

Moderate harm:
5. Append-mode topics used for auditing or compliance logging Here duplicates are visible but don't cause incorrect aggregations — they inflate counts. Annoying but not catastrophic.
6. Topics consumed by systems that deduplicate themselves If the consumer is a database with its own UPSERT semantics keyed on the business key, it will naturally collapse the duplicates. The harm is wasted I/O and potential transient inconsistency windows, not permanent data corruption.

Low or no harm:
7. Pure event streams where every event is intentionally unique If you're publishing click events or sensor readings and every record is genuinely a new fact (append mode, no deduplication intent), partition distribution is irrelevant to correctness.
8. Topics consumed only by systems that read all partitions and deduplicate at query time If your entire consumption layer uses the ROW_NUMBER() view pattern described in the document, partition-level duplicates are invisible to end users. The cost is query performance, not correctness.

The Meta-Point
The underlying issue is that Kafka's partition is a routing/ordering unit, not a semantic unit — but Tableflow's upsert mode necessarily promotes it to a semantic unit (part of the primary key) because that's the only stable identity Kafka itself provides. This creates a fundamental tension: Kafka was designed to allow flexible, independent producer behavior, while Tableflow upsert requires a global invariant (
──────────────────────
Q: What is it about the debezium envelope that might cause an issue for key + partition deduping?

The Debezium Envelope and Key/Partition Deduplication
This is a subtle but important issue. Let me break it down carefully.

What the Debezium Envelope Actually Is
When Debezium captures a change event, it doesn't just emit the row's current state. It wraps it in a structured envelope that looks something like this:
{
  "before": { "id": 1001, "email": "old@example.com" },
  "after":  { "id": 1001, "email": "new@example.com" },
  "op":     "u",
  "source": { "lsn": 12345678, "ts_ms": 1700000000000, "table": "customers" },
  "ts_ms":  1700000000123
}
The Kafka message key, by contrast, is typically just the primary key fields — e.g., {"id": 1001}.

Why This Creates a Problem for Deduplication
Here's the core tension: Tableflow's upsert deduplication works on the Kafka message key and partition. That part is fine — the key is still {"id": 1001}, so two updates to the same row will correctly overwrite each other in the Iceberg table.
The problem isn't with Tableflow's deduplication mechanism per se. The problem is with what ends up in the table and how you reason about identity afterward:

1. The Table Schema Reflects the Envelope, Not the Row
If you don't unwrap the envelope, the Iceberg table that Tableflow materializes has columns like before, after, op, source, tsms — not id, email, updatedat. So when you or a downstream analyst tries to deduplicate or identify "the current state of customer 1001," they have to write:
WHERE after.id = 1001
...instead of just WHERE id = 1001. This is error-prone and easy to get silently wrong — a filter on a top-level business column simply misses rows rather than throwing an error.

2. The op='d' Row Is the Ghost Problem
This is the more insidious deduplication-adjacent issue. When Debezium captures a DELETE, it emits:
{ "before": { "id": 1001, ... }, "after": null, "op": "d" }
This is not a Kafka tombstone (a message with a null value). It's a real message with a real value. Tableflow's upsert mode only removes a row from the Iceberg table when it sees a true tombstone (null value). So this op='d' record upserts a row into your table — a row that says "this entity was deleted." The original row is overwritten (good), but you now have a ghost row representing the deletion event itself sitting in your table.
If the source later re-inserts that primary key, Debezium emits a new op='c' record. Now you have the new row — but depending on timing, compaction state, and reader behavior, analysts may see both the ghost delete row and the new insert row as "duplicates" even though Tableflow handled the keys correctly.

3. Key Semantics Can Shift If the Envelope Affects Key Extraction
If someone configures Debezium to use a non-default key serialization — for instance, using the full envelope as the key rather than just the extracted PK fields, or switching from Avro to JSON key format mid-stream — then the key bytes change. Tableflow compares raw serialized key bytes. So {"id":1001} in Avro and {"id":1001} in JSON are different keys to Tableflow, meaning the old row is never overwritten and you get a genuine duplicate row in the Iceberg table.

4. Snapshot + Streaming Overlap Compounds This
During a Debezium snapshot, records are emitted with op='r' (read). When streaming begins, the same rows may come through again as op='c' (create) or op='u'. If the key serialization or envelope structure differs between snapshot and streaming phases (which can happen with certain connector configurations or version upgrades), those records land as distinct keys — orphaning the snapshot rows permanently.

The Fix
The document points to two remedies:
after.state.only = true on the Debezium connector — this makes Debezium emit only the after state as the message value, making it look like a normal application-produced change stream. No envelope, no before, no op. The table schema becomes clean.
Unwrap with Flink — consume the raw envelope topic, run it through a Flink job with changelog.mode = 'upsert' and the ExtractNewRecordState SMT equivalent, and write to a new clean topic that Tableflow then reads. This gives you full control and lets you also handle op='d' by emitting proper tombstones.
The key insight is: Tableflow's key+partition deduplication is working correctly even with the envelope — the envelope doesn't break the mechanism, it breaks the semantics and the schema, making it very easy to write queries that appear to deduplicate but are actually operating on the wrong level of abstraction.
──────────────────────
Q: Help me intuit this solution to normalize Kafka messages using Flink

Intuiting the Flink Re-keying Solution
Let me build this up from first principles so the mechanics click rather than just being steps to follow.

The Core Problem It's Solving
Tableflow's upsert mode is essentially saying: "I will keep the latest record per (key, partition) pair." That's its contract. The problem is that Kafka's partition assignment is determined by the producer, and producers can be inconsistent — different libraries, partition count changes, explicit partition overrides, etc. So the same logical entity (say, customer_id = 42) can end up in partition 0 and partition 3, and Tableflow dutifully keeps both, because from its perspective they're different rows.
The insight is: Tableflow can't fix this because it doesn't own a repartitioning step. It only sees what Kafka hands it. So you need to fix it before Tableflow sees it — and Flink is the right tool because it can read a topic, reason about the data, and write a new topic with a controlled, consistent key and partition assignment.

The Mental Model: A "Normalization Layer"
Think of it like a database normalization step, but for a stream:
Raw Topic (messy)          Flink Job              Clean Topic         Tableflow
─────────────────    →    (re-key + dedup)    →   (normalized)    →   (Iceberg table)
- inconsistent keys                               - stable keys
- multiple partitions                             - consistent partitioning
  for same entity                                - one row per entity
- maybe duplicates
Flink becomes your single source of truth for key→partition mapping. Once you own that mapping in Flink, everything downstream is clean.

Why changelog.mode = 'upsert' Is the Key Ingredient
When you write to a Kafka topic from Flink SQL with changelog.mode = 'upsert', you're telling Flink:
"This topic represents the current state of a keyed table. For each key, only the latest value matters. Emit upsert messages (insert/update = record with value, delete = tombstone with null value)."
This has two important effects:
Flink hash-distributes on the primary key — deterministically. customerid = 42 will always go to the same partition, regardless of which Flink task writes it, because the partition is computed as hash(primarykey) % num_partitions. No more partition drift.
The topic becomes semantically clean — it's no longer "a stream of events that happened to involve customer 42" but rather "the current state of customer 42." Tableflow then reads this clean contract.
Compare this to changelog.mode = 'append' where Flink just emits every row as a new record — no deduplication, no key-based routing guarantee.

Walking Through the Actual Mechanics
Step 1: The Re-keying Statement
INSERT INTO clean_customers_topic   -- sink: upsert-kafka connector, PK = customer_id
SELECT customer_id, name, email, updated_at
FROM raw_customers_topic;           -- source: whatever messy topic
Behind the scenes, Flink:
Reads every record from the raw topic
Routes each record to a partition based on hash(customer_id) % partitions
Writes it as an upsert message
Now customer_id = 42 is always in partition hash(42) % N. Stable. Predictable.

Step 2: The ROW_NUMBER() Dedup (When You Need It)
The re-keying above handles the partition drift problem. But what if you also have genuine duplicates — multiple records with different keys that represent the same logical entity? (e.g., a retry that minted a new UUID, or an event-keyed topic where you want latest-state-per-entity)
INSERT INTO clean_orders_topic
SELECT order_id, customer_id, amount, status, updated_at
FROM (
  SELECT *,
         ROW_NUMBER() OVER (
           PARTITION BY order_id        -- business key
           ORDER BY updated_at DESC     -- keep the latest
         ) AS rn
  FROM raw_orders_topic
)
WHERE rn = 1;
This is Flink's streaming deduplication pattern. Here's what makes it work in a streaming context (which is non-obvious):
Flink maintains state for each order_id — it remembers what it's seen
When a new record arrives for order_id = 99, Flink compares it to the stored state
If the new record is "later" (higher updated_at), it emits an upsert (update) to the output topic
If it's "earlier" (out-of-order arrival), it either ignores it or emits a retraction, depending on configuration
This is fundamentally different from a batch ROW_NUMBER() — the streaming version is incremental and stateful.

The State Size and TTL Warning — Why It Matters
The document says "budget for state size and set a TTL" — this deserves elaboration because it's a real operational concern.
For every unique order_id Flink has ever seen, it keeps a record in its state backend (RocksDB in production). If you have:
100M unique orders
Each state entry ~500 bytes
That's ~50GB of state
This is fine if you size your Flink cluster appropriately. But the real danger is unbounded state growth — old orders from 3 years ago are still in state, taking up space, even though you'll never get a late-arriving update for them.
The fix is a state TTL:
-- In Flink SQL
SET 'table.exec.state.ttl' = '7 days';
This tells Flink: "If I haven't seen an update for a given key in 7 days, evict it from state." The tradeoff: if a genuinely late record arrives after the TTL, Flink will treat it as a new entity rather than an update. For most use cases, that's acceptable — you tune the TTL to be longer than your realistic late-arrival window.

The "Two Topic" Architecture and Why It's Worth the Cost
The document acknowledges this "costs a compute pool and a second topic." Here's why that cost is often worth it:
| Concern | Without Re-keying | With Re-keying | |---|---|---| | Partition drift | Any partition count change breaks dedup | Flink controls partitioning; it's stable | | Multiple producers | Each producer can hash differently | One Flink job is the single writer | | Dedup logic location | Replicated in every query/view | Centralized once in Flink | | Downstream consumers | All must handle duplicates | All get a clean table | | Tableflow contract | (key, partition) — fragile | business_key — explicit |
The second topic is your "clean" or "silver" layer in a medallion architecture. The raw topic is bronze — it captures everything faithfully. The Flink-normalized topic is silver — it enforces the contract. Tableflow sits on silver, not bronze.

A Concrete Intuition Pump
Imagine a post office (Kafka) with 8 sorting bins (partitions). Multiple mail carriers (producers) are dropping off letters for "John Smith at 42 Oak Street." One carrier puts it in bin 3, another puts it in bin 7 (different routing logic). The post office keeps both letters because they're in different bins.
Flink is like hiring a single, careful mail sorter who:
Reads all incoming mail
Always puts "John Smith at 42 Oak Street" in exactly bin hash("42 Oak Street") % 8 — always bin 5, say
If two letters arrive for John, keeps only the most recent one
Now Tableflow reads from this sorted mailroom and sees exactly one letter per address. Clean.

When to Use This vs. the Read-Time View
The document offers read-time dedup (a ROW_NUMBER() view) as a lighter alternative. Here's the honest tradeoff:
Use the Flink re-key approach when:
Multiple downstream consumers need clean data (you pay the dedup cost once, not per query)
You have partition drift (the view can't fix this — it's a structural problem)
You need clean data for

──────────────────────
Q: What would be the best way conceptually (pseudocode) to identify duplicates in a topic, specifically the outdated record using Flink

Identifying Outdated Duplicate Records in a Kafka Topic Using Flink
This is a great problem to think through carefully, because the challenge isn't just "find duplicates" — it's "find the stale copy after a partition-count increase," which means you need to reason about which record is the authoritative survivor and which is the orphan.

The Core Mental Model First
After a partition increase, you have this situation:
Partition 0 (old):  key="order-123" → row written at t=100
Partition 2 (new):  key="order-123" → row written at t=150  ← producer now routes here
Tableflow sees these as two different primary keys (because its PK is kafkakey + kafkapartition). So both rows survive. The "outdated" record is the one in the old partition — it's the one that will no longer receive updates, because the producer has drifted to the new partition.
The authoritative record is the one in the partition where the most recent write landed.

Conceptual Approach
The algorithm has three stages:

Stage 1: Group by Business Key, Find the Winning Partition
FOR EACH message in the topic:
    emit (business_key, partition, offset, event_timestamp, full_record)

GROUP BY business_key:
    winning_partition = partition where MAX(event_timestamp) was seen
                        -- tie-break: MAX(offset) within that partition
                        -- tie-break-2: MAX(partition number) as last resort
The winning partition is the one the producer is currently routing to. Any other partition that also holds this business key is stale.

Stage 2: Identify Stale (partition, key) Pairs
FOR EACH business_key with records in multiple partitions:
    stale_records = all records WHERE partition != winning_partition
    emit (stale_record.original_kafka_partition, stale_record.kafka_key)
These are your tombstone targets.

Stage 3: Emit Tombstones to the Stale Partitions
FOR EACH (stale_partition, kafka_key) in stale_records:
    produce tombstone:
        topic     = original_topic
        key       = kafka_key          -- exact original bytes, not re-serialized
        value     = null               -- true tombstone
        partition = stale_partition    -- MUST target the original partition explicitly
Flink SQL Pseudocode
-- Step 1: Read the topic as a table, exposing metadata
-- Flink can expose kafka partition and offset as metadata columns

CREATE TABLE kafka_source (
    kafka_key       BYTES METADATA FROM 'key',
    business_key    STRING,           -- extracted from key or value
    event_ts        TIMESTAMP(3),     -- source system timestamp (LSN/SCN preferred)
    kafka_partition INT METADATA FROM 'partition',
    kafka_offset    BIGINT METADATA FROM 'offset',
    payload         ROW<...>
) WITH (
    'connector' = 'kafka',
    'topic' = 'your-topic',
    'scan.startup.mode' = 'earliest-offset',
    ...
);

-- Step 2: For each business key, find the winning partition
-- (the one that received the most recent write)

CREATE VIEW winning_partitions AS
SELECT
    business_key,
    -- The partition that has the latest event is the authoritative one
    LAST_VALUE(kafka_partition) OVER (
        PARTITION BY business_key
        ORDER BY event_ts, kafka_offset   -- offset as tiebreaker
    ) AS winning_partition,
    MAX(event_ts) OVER (PARTITION BY business_key) AS latest_ts
FROM kafka_source;

-- Step 3: Identify stale records
-- These are rows where this record's partition is NOT the winning partition
-- AND there exists a newer record for the same business_key in a different partition

CREATE VIEW stale_records AS
SELECT DISTINCT
    ks.kafka_key,
    ks.kafka_partition   AS stale_partition,
    ks.business_key
FROM kafka_source ks
JOIN winning_partitions wp
    ON ks.business_key = wp.business_key
WHERE ks.kafka_partition <> wp.winning_partition
  AND ks.event_ts < wp.latest_ts;   -- confirm this partition is truly behind

-- Step 4: Write tombstones back to the original partitions
-- (Flink DataStream API for explicit partition targeting)

INSERT INTO kafka_sink  -- configured with fixed-partition routing
SELECT
    kafka_key,
    stale_partition,
    CAST(NULL AS BYTES) AS value   -- tombstone
FROM stale_records;
Important Nuances You Must Handle
1. Tombstone Delivery Requires Explicit Partition Assignment
This is critical. You cannot let the partitioner decide where to send the tombstone — it will route to the new partition (which is exactly the problem). In Flink's DataStream API:
# Pseudocode for the producer side
producer.send(
    topic     = original_topic,
    partition = stale_partition,    # HARDCODED to the old partition
    key       = original_key_bytes, # exact bytes, no re-serialization
    value     = None                # tombstone
)
In Flink SQL, you'd need a custom sink or use the kafka connector with a partitioner configured to pass through an explicit partition column.

2. Key Bytes Must Be Exact
If the original key was Avro-serialized, your tombstone key must also be Avro-serialized identically. Deserializing to a string and re-serializing will produce a different byte sequence and Tableflow/log compaction won't match it.
-- WRONG: deserialize then re-serialize
tombstone_key = avro_serialize(avro_deserialize(original_key))  # may differ

-- RIGHT: pass through the raw bytes
tombstone_key = original_key_bytes  # untouched
3. Ordering Problem in Streaming Mode
In a streaming context, you can't know a partition is "stale" until you've seen a later write in another partition. This means:
Batch mode (scan full topic from earliest): straightforward, process all records then identify staleness
Streaming mode: you need a watermark + a time window, and you accept some latency before declaring a partition stale
For this use case, batch mode is almost always correct — you're doing a one-time remediation after a partition-count change, not continuous dedup.

4. The Diagnostic Query First
Before writing tombstones, run the diagnostic to confirm your assumptions:
SELECT
    business_key,
    COUNT(*)                        AS total_rows,
    COUNT(DISTINCT kafka_partition) AS partition_count,
    MIN(event_ts)                   AS oldest_event,
    MAX(event_ts)                   AS newest_event,
    COLLECT(kafka_partition)        AS partitions_seen
FROM kafka_source
GROUP BY business_key
HAVING COUNT(DISTINCT kafka_partition) > 1
ORDER BY total_rows DESC
LIMIT 100;
If partition_count > 1, you have partition drift. This is your confirmation before you start emitting tombstones.

The Full Conceptual Flow
kafka_source (full scan, earliest offset)
    │
    ▼
[Group by business_key]
    │
    ├─→ Find winning_partition (latest event_ts + offset tiebreaker)
    │
    ├─→ Identify stale (partition, key_bytes) pairs
    │       where partition != winning_partition
    │
    └─→ Emit null-value tombstone to each stale partition
            with EXACT original key bytes
            targeting EXACT original partition number
                │
                ▼
        Kafka log compaction drops stale record
        Tableflow drops stale row on next compaction cycle
One Gotcha Worth Calling Out
If the same business key has been through multiple partition increases (e.g., 3 → 6 → 12), you may have records stranded in several old partitions. The algorithm above handles this naturally — all partitions except the winning one get tombstones — but make sure your scan goes back to `earliest-offset

Unstarred

Dunk (exclude from study)

Duplicate Records in Tableflow

Review Questions

Todos

Sensemaker