kuni-92 opened a new issue, #18623:
URL: https://github.com/apache/iceberg/issues/18623
**Apache Iceberg version:** 1.12.0 (reproduced). The code involved appears
unchanged on `main` (`flink/v2.2`), so the same behaviour is expected there.
**Query engine:** Flink (`DynamicIcebergSink`, DataStream API, upsert mode)
**Please describe the bug 🐞**
### Summary
`DynamicWriter` and `DynamicWriteResultAggregator` cache per-table objects
(task writer factories / manifest output file factories) that capture the
`Table` loaded when the entry was created. The caches are keyed by table name
(plus schema id / spec id for the writer), have no TTL and are never
invalidated. When a table is dropped and re-created with the same name while
the job is running, the cache key can match the old entry, so the sink keeps
writing data files and manifests for the **dropped** table's location.
On Amazon S3 Tables this fails at the next checkpoint with
`AccessDeniedException (403) Could not access through this access point`,
because the dropped table's storage is no longer accessible. The TaskManager
fails, the job restarts, the caches are rebuilt from scratch and the replayed
records are written to the new table.
### Root cause
Line numbers at tag `apache-iceberg-1.12.0`, `flink/v2.2`.
1. `DynamicWriter#taskWriterFactories` is an `LRUCache<WriteTarget,
RowDataTaskWriterFactory>` (`DynamicWriter.java:59,82`). `WriteTarget` consists
of table name, branch, schema id, spec id, upsert mode and equality fields
(`DynamicWriter.java:93-99`), with no table identity such as the table UUID.
2. On a cache miss the factory loads the table once and captures it as `()
-> table` (`DynamicWriter.java:105-106,133-134`). The `writers` map is cleared
after every `prepareCommit()` (`DynamicWriter.java:198`), but the factories
stay in the cache, so new writers in later checkpoints reuse the stale `Table`
(location, `FileIO`).
3. `DynamicWriteResultAggregator#outputFileFactoriesAndFormatVersions` and
`#specs` are `LRUCache`s keyed by table name only
(`DynamicWriteResultAggregator.java:71-72,99-100`).
`outputFileFactoryAndFormatVersion()` captures the loaded table the same way
(`DynamicWriteResultAggregator.java:210-230`), so manifests are also created
for the stale table.
4. `TableMetadataCache` (used by `DynamicRecordProcessor`) refreshes after
`cacheRefreshMs` and sees the re-created table, so records reach the writer
with the new table's schema id / spec id. A newly created table starts again at
schema id 0 / spec id 0, which hits the stale `WriteTarget` entry.
### Reproduction
DataStream job with `DynamicIcebergSink` (REST catalog on Amazon S3 Tables,
upsert mode, equality field `id`, `immediateTableUpdate(true)`):
1. Write some rows to table `db.t` (schema `(id INT PK, val STRING)`) and
let a checkpoint commit them.
2. Drop the table from the catalog (`dropTable(db.t, purge=true)`) while the
job keeps running.
3. Write a row for `db.t` again, with the same or a different schema. The
table is re-created (schema id 0, spec id 0).
4. At the next checkpoint:
```
Sink: Writer ... switched from RUNNING to FAILED
java.io.IOException: Could not perform checkpoint N for operator Sink: Writer
Caused by: java.io.IOException:
software.amazon.awssdk.services.s3.model.AccessDeniedException: Could not
access through this access point (Service: S3, Status Code: 403, ...)
at
org.apache.iceberg.parquet.ParquetWriter.close(ParquetWriter.java:285)
at org.apache.iceberg.io.DataWriter.close(DataWriter.java:83)
at
org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.closeCurrent(BaseTaskWriter.java:435)
...
at
org.apache.iceberg.flink.sink.UnpartitionedDeltaWriter.close(UnpartitionedDeltaWriter.java:69)
at
org.apache.iceberg.io.BaseTaskWriter.complete(BaseTaskWriter.java:148)
```
**Expected**: The sink writes to the re-created table.
**Actual**: The writer writes to the dropped table's location and the
checkpoint fails. After the restart the job recovers, because the caches are
empty.
### Impact
- On Amazon S3 Tables, every same-name drop + re-create causes one job
failure / restart.
- CDC pipelines that replicate `DROP TABLE` / `CREATE TABLE` (e.g. Debezium)
hit this whenever a table is re-created with the same name.
### Proposed fix
Either or both:
1. Include the table identity (table UUID) in the cache keys of
`DynamicWriter` and `DynamicWriteResultAggregator`, e.g. by carrying the UUID
resolved by `TableMetadataCache` in `DynamicRecordInternal` /
`DynamicWriteResult`.
2. Validate the cached `Table` when a new writer is created for a checkpoint
(e.g. compare `table.uuid()` with a freshly loaded / `TableMetadataCache`
value), and rebuild the factory on mismatch. Optionally bound the cache entries
by `cacheRefreshMs` like `TableMetadataCache`.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [x] I cannot contribute a fix for this bug 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]
