dkranchii opened a new pull request, #17819: URL: https://github.com/apache/iceberg/pull/17819
## Which Iceberg project does this PR belong to? Kafka Connect ## What changes are proposed in this pull request? Fixes #17641. Under dynamic routing, when the sink is configured with `iceberg.tables.auto-create-enabled=false` and a record is routed to a table that does not exist in the catalog, `IcebergWriterFactory.createWriter` returned a `NoOpWriter` that silently discarded every subsequent record for that table. There was no log line, no counter, and no metric — the connector stayed in `RUNNING` state while data was lost. The reporter documented two AWS MSK Connect production incidents in the issue and proposed the fix. This PR: - Adds a `Set<String> warnedMissingTables` (backed by `ConcurrentHashMap.newKeySet()`) to `IcebergWriterFactory`. The first time a given `tableName` resolves to `NoOpWriter`, the factory logs a WARN that includes the table name, the catalog name, and the actionable hint from the issue. Subsequent misses for the same table do not re-log (log-flood protection at high record throughput). - Adds a factory-scoped `AtomicLong droppedRecordCount`. `NoOpWriter` now accepts a reference to this counter in its constructor and increments it on every `write(record)`. This gives operators the precise number of records lost, not just the number of unique unresolved tables. - Exposes both fields via `@VisibleForTesting` accessors so unit tests can assert the behavior deterministically without needing an slf4j appender fixture. Behavior for the happy path (table exists, or `ignoreMissingTable=false`) is unchanged. ### Design choices vs. reporter's open questions - **Q1 — Metric name.** Not addressed in this PR. Wiring the counter into a Kafka Connect sink metric is deferred so the metric name can be discussed independently. - **Q2 — WARN wording.** Matches the reporter's pred message in the issue. - **Q3 — Rate-limit shape.** Log-once-per-table-name-per-factory, matching the reporter's simplest proposal. - **Q4 — Scope of the counter.** Single factory-scoped counter (task-scoped in practice, since each `SinkWriter` has one factory per task). Per-table breakdown is deferred to keep the diff minimal. - **Q5 — Where to increment.** Incremented on `NoOpWriter.write(record)` so the counter reflects records lost, not unique missing tables. ## Tests New unit tests in `TestIcebergWriterFactory`: - `missingTableWithIgnoreFlagReturnsNoOpWriterAndTracksTable` — the first miss returns a `NoOpWriter` and adds the table to `warnedMissingTables`. - `missingTableRecordsWarnAtMostOncePerTable` — two calls for the same missing table produce exactly one entry (proxy for "WARN logged once"). - `missingTablesEachRecordSeparateWarn` — two different missing tables each get their own warned-set entry. - `noOpWriterIncrementsDroppedRecordCountPerWrite` — `NoOpWriter.write` incrementsunter once per call. - `droppedRecordCountAccumulatesAcrossTables` — the counter is factory-scoped and accumulates across all `NoOpWriter` instances the factory returns. - `missingTableWithoutIgnoreFlagRethrows` — regression: when `ignoreMissingTable=false`, `NoSuchTableException` still propagates and no warn / counter increment happens. Existing `testAutoCreateTable` continues to pass unchanged. Existing `TestSinkWriter` tests (which construct `IcebergWriterFactory` indirectly through `SinkWriter`) also continue to pass — the change is behavior-additive. --- **AI Disclosure** - Platform/Tool: Cursor -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
