apache / apache/iceberg

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

Open
#17,542 2 comments 3 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 11h
Merged PRs (30d)
132

Description

## Motivation

The pre-donation project ([tabular-io/iceberg-kafka-connect](https://github.com/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..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/#15046 itself start working again, or keep our per-table shape (`iceberg.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](https://iceberg.apache.org/contribute/#guidelines-for-ai-assisted-contributions).

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

Contributor guide

Open the contributing guide

Research direction

Start with the proposed branch and the Kafka Connect module's existing InMemoryCatalog/WriterTestBase JUnit suite, then review the stated config, writer, tombstone, topic-routing, partition-spec, test, and documentation changes. Done means maintainer alignment on the five open design questions followed by a focused, reviewable PR with the behavior and tests agreed by the project.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, kafka
Domain
data-engineering, databases, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
28/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.