Kafka Connect: log and count records routed to a non-existent table under dynamic routing (NoOpWriter currently drops them silently)

## Motivation

When the sink runs in dynamic routing mode (`iceberg.tables.dynamic-enabled=true`), the destination table for each record is derived at write time from the record's route field. If the derived table doesn't exist in the catalog **and** the connector is configured with `iceberg.tables.auto-create-enabled=false`, the sink silently discards every such record via `NoOpWriter`.

The drop happens in `IcebergWriterFactory.createWriter`:

```java
} else if (ignoreMissingTable) {
return new NoOpWriter();
}
```

No log, no counter, no metric. `NoOpWriter.write(record)` is a genuine no-op — the record disappears with no diagnostic surface. The caller in `SinkWriter.routeRecordDynamically` passes `ignoreMissingTable=true` unconditionally, so this branch is the default fallback whenever dynamic routing encounters a name the catalog doesn't yet have.

This produces exactly the same *"connector is RUNNING, no data in the target table, no ERROR in logs"* symptom class as several other issues (missing IAM permissions, wrong control topic, misconfigured `cdc-field`, `schema-force-optional` colliding with id-columns). It is arguably the hardest of that group to diagnose because there is no stack trace anywhere — the record's journey ends inside a class named `NoOpWriter`, which behaves as advertised. Operators typically only realize the drop is happening after enabling TRACE-level logging on a specific record path or after correlating Kafka consumer-group lag (records consumed) against Iceberg row counts (records visible).

## When this fires in practice

Not just the obvious "auto-create off, table missing" case. Several deployment shapes hit it:

- **Auto-create is off by policy** — governance requires tables to be pre-created via IaC / DDL. A dynamic-routing connector then receives a record for a table that hasn't been pre-created yet (Debezium picked up a new source table, upstream added a shard). Every such record drops.

- **Copy-pasted config carrying `dynamic-enabled=true` from a sibling connector** without `auto-create-enabled=true`. Everything else in the config is right; every record silently drops until the operator notices row counts aren't advancing.

- **A new source table name appears mid-stream after auto-create was manually disabled** — common when an operator toggles `auto-create-enabled=false` on a running dynamic-routing connector to prevent further table sprawl, but doesn't restart. Existing writers keep working; any new table name in the stream silently no-ops.

## What we're proposing

Emit a single **WARN** log line the first time a given table name resolves to `NoOpWriter` per task lifetime, and increment a task-scoped drop counter. Concretely:

- Change the return path in `IcebergWriterFactory.createWriter` so that when the fallback returns `NoOpWriter`, the factory logs at WARN with the specific table name, the catalog identifier, and a hint (*"Records for this table will be silently discarded. Set `iceberg.tables.auto-create-enabled=true` or pre-create the table in the catalog."*).
- Track the set of already-warned tables in the factory so the WARN doesn't repeat on every subsequent record (avoids log flooding under high throughput).
- Increment a task-scoped counter every time a `NoOpWriter.write(record)` is called (not just on writer creation), so operators can distinguish "one bad name, one warning" from "10 M records lost." The counter is surfaced through a Kafka Connect sink metric (name to be agreed — see open questions).

The Iceberg sink currently emits no Kafka Connect metrics of its own beyond what the framework provides. This proposal would be the first sink-scoped metric; the pattern established here could be reused for other proposed drop-counters (record-drop on missing route field, commit failures, etc.).

Behavior for the correctly-configured happy path is unchanged.

## What we've verified in production

We're running the sink on AWS MSK Connect with Glue + S3FileIO. We hit this twice:

- Once during a Debezium re-parenting: the source connector started emitting to a new topic whose target table hadn't been pre-created. The Iceberg sink ran without complaint; row counts never advanced. Diagnosis required 40 minutes and reading the sink source to find the `NoOpWriter` branch.
- Once when an operator disabled `auto-create-enabled` as part of a security-review change without realising a running dynamic-routing connector was silently relying on it. No records reached the tables that had been auto-created previously and were then dropped by a separate cleanup — but the connector's status remained `RUNNING` with no error signal.

In both cases the fix on the operator side was trivial once the drop was identified. The gap is entirely on the diagnostic side.

Not something we've prototyped in a branch — the fix is small enough (~30 lines main + ~40 lines test) that filing here first for wording and metric-name alignment is cheaper than iterating on a PR.

## Open questions for maintainers, before a PR

1. **Metric name.** Proposed `iceberg-sink:records-dropped-no-table`, mirroring the shape of existing Kafka Connect sink metrics. Alternatives: `iceberg-sink:noop-writes`, `iceberg-sink:records-discarded`. Preference?

2. **WARN wording.** Proposed:

> Table `` does not exist and auto-create is disabled; records routed to it will be silently discarded. Set `iceberg.tables.auto-create-enabled=true` or pre-create the table in the catalog.

Anything you'd tighten?

3. **Rate-limit shape.** Log-once-per-table-name-per-task is simple and covers the diagnostic case. Alternative: log on every N-th drop. Preference?

4. **Scope of the counter.** Task-scoped counter is the simplest; per-table breakdown would be more actionable but requires managing a `Map` that grows with unique unresolved names. Worth the complexity, or start with a single counter and defer per-table?

5. **Should `NoOpWriter` itself emit the counter increment, or should the factory increment on writer creation only?** Incrementing on `write` gives an accurate record-count but requires threading the counter into every `NoOpWriter` instance. Creating-time increment is simpler but only tells you "how many unique tables missed," not "how many records were lost."

## Contribution

Happy to open a PR against `main` with the WARN log, the metric, and unit tests covering:

- First-drop per table logs at WARN; subsequent drops don't.
- `NoOpWriter.write` increments the counter (or factory-time increment, depending on outcome of Q5).
- Happy path (table exists, auto-create-enabled works) is unaffected — no WARN, no counter increment.

This proposal was drafted with AI assistance (Claude); the production evidence and design decisions have been reviewed and validated by us against real MSK Connect deployments, per the project's [[AI-assisted contribution guidelines](https://iceberg.apache.org/contribute/#guidelines-for-ai-assisted-contributions)](https://iceberg.apache.org/contribute/#guidelines-for-ai-assisted-contributions).

### Query engine

Kafka Connect

### Willingness to contribute

- [x] I can contribute this improvement/feature independently
- [ ] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time

Start with IcebergWriterFactory.createWriter, SinkWriter.routeRecordDynamically, and NoOpWriter.write to trace the missing-table fallback. Review the project's existing Kafka Connect metric and logging patterns before resolving the open questions about metric naming, warning rate limiting, and counter ownership. Done means covering first and subsequent drops plus the unaffected happy path with unit tests.

Written by the indexing model from the issue text.