apache / apache/iceberg

Kafka Connect: id-columns config ignored for namespaced tables

Open Beginner friendly
#17,325 2 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.