FizzaAbid opened a new issue, #17641:
URL: https://github.com/apache/iceberg/issues/17641
### Feature Request / Improvement
## 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 `<namespace.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<TableName,
Counter>` 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
- [ ] I can contribute this improvement/feature independently
- [x] I would be willing to contribute this improvement/feature with
guidance from the Iceberg community
- [ ] I cannot contribute this improvement/feature at this time
--
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]
