Kafka Connect: real CDC upsert/delete support (equality-deletes) - restoring and fixing pre-donation cdc-field/upsert functionality

[raphaelpfl]

Motivation

The pre-donation project (tabular-io/iceberg-kafka-connect, now archived) advertised "Row mutations (update/delete rows), upsert mode" as a feature, configured via iceberg.tables.cdc-field (op codes I/U/D) and iceberg.tables.upsert-mode-enabled. Neither made it into the code donated to apache/iceberg - the current kafka-connect module is append-only.

Users are still hitting this gap with the exact old config keys, which are now silently ignored (unknown connector properties): #15046. Both the OP and a commenter report duplicate rows in Iceberg even with upsert-mode-enabled/cdc-field set, using a real Debezium source.

Digging into why it wouldn't have worked even if those keys were still wired up: the old CDC docs only ever show a value that carries the op code inline (e.g. _cdc_op: "I"/"U"/"D") - never a delete event with a null value. In practice, with a Debezium source and the ExtractNewRecordState SMT, delete events arrive at the sink as a standard Kafka tombstone (null value, key only) regardless of delete.handling.mode - not as a rewritten row carrying an op field. So the original design would not have handled a real Debezium delete correctly even if it had survived the donation.

What we prototyped and validated

We (Junto Seguros) built a fix on top of apache-iceberg-1.11.0 and tested it end-to-end against a real pipeline: Debezium (SQL Server) -> Kafka -> this sink -> Iceberg (REST catalog / AWS S3 Tables). We've since rebased it cleanly onto current main (no conflicts touching any of the files below). Summary:

• cdc-field (iceberg.table.<table>.cdc-field, with a connector-wide default iceberg.tables.default-cdc-field): the named record-value field is compared against Debezium op codes c/r (insert), u (update: delete then insert), d (delete). Backed by Iceberg equality-deletes (BaseEqualityDeltaWriter), not a table rewrite.
• Delete via tombstone: a null-value record, with cdc-field enabled and a non-null key, is treated as delete-by-key. This is the gap described above - it works regardless of delete.handling.mode and isn't Debezium-specific (any CDC source emitting standard tombstones on delete works).
• Partitioned tables: a key-only delete can't compute a partition value (no full row available), so it's written as an equality-delete under an unpartitioned spec from the table's spec history; Iceberg applies it against every partition on read. Tables auto-created by the connector always get such a spec for free (created unpartitioned, then evolved). A pre-existing partitioned table without that history can opt into a one-time spec-evolution migration (cdc-auto-migrate-partition-spec) that adds one without changing how future writes are partitioned.
• Topic-based routing (route-by-topic-enabled + dynamic-topic-namespace + topic-table-regex): routes by Kafka topic name instead of a record-value field, so it works correctly for null-value tombstones (unlike dynamic-enabled, which inspects the value and would silently drop every delete) and lets one connector instance fan out to hundreds/thousands of source tables (one topic per table, Debezium-style naming) without a connector-per-table topology.

All of this is covered by a JUnit suite (161 tests, including the module's pre-existing suite) using the module's existing InMemoryCatalog/WriterTestBase conventions, plus manual end-to-end validation against a live pipeline (including a schema-evolution-during-CDC scenario).

Working branch, rebased onto current main, split into small reviewable commits (config -> writer -> wiring -> topic routing -> partitioned-table support -> docs): https://github.com/raphaelpfl/iceberg/tree/cdc-upsert-delete-support - posted for early visibility, not as a ready-to-merge PR; opening the PR itself once there's some directional alignment on the questions below.

Open questions for maintainers, before a PR

A few things affect public config surface and would benefit from maintainer input before we invest in a polished PR:

1. Backward-compat naming: reuse the exact pre-donation key (iceberg.tables.cdc-field, connector-wide) so existing tutorials/blog posts/Does kafka connector it support upsert #15046 itself start working again, or keep our per-table shape (iceberg.table.<table>.cdc-field + iceberg.tables.default-cdc-field)? We can support both if that's preferred.
2. Op code casing: the old design used uppercase I/U/D; our prototype matches Debezium's lowercase c/r/u/d, since that's what we tested against. Should this be configurable (an op-code mapping), fixed to one convention, or support a couple of well-known conventions out of the box?
3. upsert-mode-enabled: the old design had this as an independent flag (equality-delete before every insert, no cdc-field needed - useful for idempotent/exactly-once-ish reprocessing without real deletes). We haven't implemented this; open to adding it as a follow-up if there's interest.
4. Topic-based routing vs. extending dynamic-enabled: is a new routing mode the right shape, or would maintainers prefer dynamic-enabled itself became null-safe (e.g. falling back to the topic name when the record value is null)?
5. cdc-auto-migrate-partition-spec: any preference between an opt-in flag (our approach, since it mutates table metadata - two partition-spec evolutions), always attempting it automatically, or leaving that kind of migration to a separate offline procedure entirely?

Contribution

We can turn this into a PR with small, reviewable commits (config -> writer -> tombstone handling -> topic routing -> partitioned-table support -> tests -> docs) once there's rough alignment on the questions above.

Part of this implementation was drafted with AI assistance (Claude); all code has been reviewed line-by-line, tested end-to-end against a real pipeline, and is covered by the JUnit suite mentioned above, per the project's AI-assisted contribution guidelines.

cc @laskoviymishka @AnatolyPopov, since you both reviewed the most recent Kafka Connect changes - would appreciate your take on the open questions above whenever you have a moment.

---

Query engine: Kafka Connect
Willingness to contribute: I would be willing to contribute this improvement/feature with guidance from the Iceberg community