Flink IcebergSink committer calls TableLoader.loadTable() per subtask, fanning out catalog loads at startup

[krisnaru]

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

• 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

[mxm]

Hi Krish! There should only be one committer instance per Flink job. If that’s not the case, we need to fix this in Flink. I'd prefer not to do a lazy init to fail fast before the first commit.

[pvary]

Hi Krish! There should only be one committer instance per Flink job. If that’s not the case, we need to fix this in Flink. I'd prefer not to do a lazy init to fail fast before the first commit.

I remember a discussion about this on the Flink dev list: https://lists.apache.org/thread/82bgvlton9olb591bfg2djv0cshj1bxj
I don't remember that there were a solution for that in Flink. So basically we instantiate multiple committers, but only use a single one (IcebergSink.addPreCommitTopology).

AI suggests that something like this should be a solution:

    @Override
    public void addPostCommitTopology(DataStream<CommittableMessage<CommT>> committables) {
        // Force Flink to funnel all committables into a single worker thread (Parallelism = 1)
        StandardSinkTopologies.addGlobalCommitter(
            committables,
            ContextualCommitterAdapter::new,  // Your committer provider
            GlobalCommittableSerializer::new  // Serializer for the aggregated committables
        );
    }

I'm not sure how this behaves along with the current topology

[mxm]

StandardSinkTopologies.addGlobalCommitter is a bit of a workaround because it doesn't replace the committer, but runs as post-commit. Under the hood, it creates an operator with parallelism 1. Unfortunately, this is only available for Flink 2.0 and above (we still support Flink 1.20). It could be a viable path.

On the other hand, the current code works fine, apart from the catalog load issue. We already aggregate all the WriteResults and send them only to a committer subtask 0. So the rest of the committer subtasks never do anything. I would suggest to just check the subtask id in the committer and skip loading the table for any subtasks id > 0. We should make sure to throw in case any messages ever make it to a subtask id > 0.

What do you think?

[pvary]

I would suggest to just check the subtask id in the committer and skip loading the table for any subtasks id > 0. We should make sure to throw in case any messages ever make it to a subtask id > 0.

Sounds reasonable to me