Kafka Connect: duplicate PK rows when two PK updates land in one snapshot after a crash
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Apache Iceberg version
_No response_
### Query engine
Kafka Connect
### Please describe the bug 🐞
## Summary
When using the Iceberg Kafka Connect sink with a primary key (equality field IDs) `(a, b, ...)`, a crash occurring between two consecutive updates of the same primary key can result in **both versions of the row living in the target table**.
The root cause is that both the old and the new versions of the row end up being committed within the **same snapshot**. Because an equality delete only removes data rows from **previous** (lower sequence number) snapshots and has no effect on data rows added in the **same** snapshot, the equality delete fails to remove the stale row. As a result, the target table ends up with duplicate rows for the same primary key.
## Scenario / Steps to reproduce
1. A record with primary key `(a, b, ...)` is updated. The sink writes the new data row plus an equality delete for the old value.
2. A crash happens before the offsets/commit are finalized, so the work is retried.
3. The same primary key `(a, b, ...)` is updated **once again**.
4. Both updates (the data rows and their equality deletes) are flushed and **committed together in a single snapshot**.
5. Query the target table.
## Expected behavior
Only the latest version of the row for primary key `(a, b, ...)` should exist in the target table.
## Actual behavior
Both versions of the row for primary key `(a, b, ...)` are present in the target table. Since both are committed in one snapshot, the equality delete has no effect on the row added in the same snapshot (equality deletes only apply to data files with a lower sequence number).
## Notes
This appears to be a consequence of equality-delete semantics: an equality delete with sequence number `S` only deletes data rows with sequence number `< S`; rows added in the same snapshot (same sequence number) are not affected. When a crash forces multiple PK updates to be batched into one snapshot, the intermediate stale version is not removed.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
Contributor guide
Research direction
The payload identifies the Kafka Connect sink but no source file or test. Start by tracing how the sink batches retries, equality deletes, and snapshot commits; reproduce the crash/update sequence and add a regression test showing that only the latest primary-key row remains.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- backend, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100