krisnaru opened a new issue, #17140:
URL: https://github.com/apache/iceberg/issues/17140
### Apache Iceberg version
1.9.2
### Query engine
Flink
### Please describe the bug 🐞
### Description
In the SinkV2-based IcebergSink, the committer operator is not pinned to
parallelism 1 — it inherits the sink/writer parallelism. Flink instantiates one
IcebergCommitter per committer subtask, and IcebergCommitter loads the table in
its constructor. A job with writer parallelism N therefore issues up to N
loadTable() calls to the catalog at startup (and on every restart/recovery).
With iceberg.parallelism = 2500 against a REST catalog, that's ~2500
concurrent loadTable requests at startup; with a large (~45–48 MB) metadata
response the burst saturates the catalog and the job fails to start:
```
org.apache.iceberg.exceptions.RESTException: Failed to convert HTTP
response body to string
at
org.apache.iceberg.rest.RESTCatalog.loadTable(RESTCatalog.java:102)
at
org.apache.iceberg.flink.TableLoader$CatalogTableLoader.loadTable(TableLoader.java:133)
at
org.apache.iceberg.flink.sink.IcebergCommitter.<init>(IcebergCommitter.java)
at
org.apache.iceberg.flink.sink.IcebergSink.createCommitter(IcebergSink.java)
(Suppressed cause: ConnectionClosedException: Premature end of chunk coded
message body.)
```
### Root cause
IcebergSink implements SupportsCommitter, so Flink calls createCommitter() →
new IcebergCommitter(...) once per subtask, and the constructor calls
tableLoader.loadTable(). Committer parallelism isn't set to 1; only the
pre-commit aggregator is (via .global() + setParallelism(1)), so only subtask 0
commits — but all N subtasks are still constructed and each loads the table.
The older FlinkSink pins its committer to parallelism 1, so this is a
regression.
Why "load once on the JM, ship serialized" doesn't apply here — that holds
for the source and the writers (which reuse the client-loaded SerializableTable
via tableSupplier), but the committer ignores it and always reloads via the
catalog.
(Full code excerpts and the FlinkSink comparison are in the file.)
Willingness to contribute
- [x] I can contribute a fix for this bug independently
### Proposed / implemented fix
Defer loadTable() to the first non-empty commit(). Since only subtask 0
receives committables (global() shuffle), only it loads the table — one load
instead of N. Backward compatible; no upstream Flink dependency. (Reusing the
SerializableTable isn't viable — it throws on
newAppend()/newRowDelta()/newReplacePartitions().)
### 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
- [ ] 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]
