Kafka Connect: id-columns config ignored for namespaced tables

[thswlsqls]

Apache Iceberg version

main @ 8550723

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.<table-name>.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().

[ebyhr]

There is an existing PR #17152

[rainerschamm]

There is an existing PR #17152

I see this PR also a test case, so lets go with this PR instead of mine...