Kafka Connect: id-columns config ignored for namespaced tables
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
**Apache Iceberg version**
main @ 8550723a7
**Query engine**
None — Kafka Connect.
**Please describe the bug**
`RecordUtils.createTableWriter()` (`.../connect/data/RecordUtils.java` line 117) looks up the per-table config with `tableReference.identifier().name()`, which drops the namespace.
`IcebergSinkConfig.tableConfig(String)` builds the prefix as `"iceberg.table." + tableName + "."`. For table `default.events` the lookup uses `iceberg.table.events.`, never finds `iceberg.table.default.events.id-columns`, and silently falls back to `iceberg.tables.default-id-columns`. With no default set, `identifierFieldIds` stays empty and an append writer is built instead of an equality delete writer — upserts accumulate duplicate rows with no error.
The other call sites reading this config map use the full name: `Coordinator.commitToTable()` line 260 (`config.tableConfig(tableIdentifier.toString())`), `IcebergWriterFactory` line 101, `SinkWriter` line 111. Only `RecordUtils` differs.
`docs/docs/kafka-connect.md` line 76 documents the key as `iceberg.table..id-columns`, with namespaced examples (lines 328-330).
**Steps to reproduce**
Set `iceberg.tables=default.events` and `iceberg.table.default.events.id-columns=id`, then send records updating an existing `id`.
Expected: equality deletes are written, rows updated.
Actual: the setting is ignored, rows appended as duplicates.
**Additional context**
Introduced by PR #14979 (commit `d85f8a87a`), which replaced the `String tableName` parameter with `TableReference` and substituted the full name with `identifier().name()`. Tests missed it because `WriterTestBase` uses an identifier without a namespace, where `name()` equals `toString()`.
Contributor guide
Research direction
Start in connect/data/RecordUtils.java at RecordUtils.createTableWriter() and compare its tableConfig lookup with Coordinator.commitToTable(), IcebergWriterFactory, and SinkWriter. Use WriterTestBase and add coverage for a namespaced identifier with iceberg.table.default.events.id-columns; done means equality deletes are selected and updates no longer append duplicate rows.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- backend, data
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 75/100