Flink: Allow independent parallelism bounds for DynamicIcebergSink committer

[Sbaia]

Feature request

Allow DynamicIcebergSink users to configure the committer's initial parallelism and maximum parallelism independently from the writer.

Versions

• Apache Iceberg: 1.11.0
• Apache Flink: 2.1.2
• Apache Flink Kubernetes Operator: 1.13.0

Problem

DynamicIcebergSink.Builder.writeParallelism(int) configures the writer. The internal pre-commit/committer topology has no independent parallelism bound and can inherit the sink parallelism plus the job-wide max parallelism.

This is problematic for autoscaled jobs because writer demand and committer demand have different semantics:

• writers process row records continuously;
• dynamic committers process checkpoint-batched metadata and distribute work by table;
• writers may legitimately need parallelism 32 while committers may only need 4 or 8.

In an observed job, the committer was scaled from 8 to 32 while its average load was 0.9%. A multi-input edge-rate estimation problem in the Flink autoscaler propagated row-scale demand into the checkpoint-batched committer vertex. There was no stable connector-level way to bound only the committer:

• job.autoscaler.vertex.exclude.ids requires a generated JobVertexID;
• a global autoscaler maximum also limits sources and writers;
• changing the job-wide Flink max parallelism can affect keyed-state compatibility.

Issue #17140 describes the related behavior in the non-dynamic SinkV2 implementation: the committer inherits writer parallelism. PR #17251 avoids unnecessary initialization in inactive subtasks, but does not provide independent committer parallelism control.

• Flink IcebergSink committer calls TableLoader.loadTable() per subtask, fanning out catalog loads at startup #17140
• Flink: Avoid unnecessary TableLoader.loadTable() in IcebergCommitter subtasks > 0 #17251

The dynamic sink should not be hard-wired to parallelism 1 because it can commit multiple tables concurrently. It should instead expose a bounded, configurable value.

Proposed API

For example:

DynamicIcebergSink.forInput(input)
    .writeParallelism(32)
    .committerParallelism(4)
    .committerMaxParallelism(8)
    .append();

The exact API may need corresponding support from the Flink SinkV2 translator, because the framework currently creates the committer operator internally.

Proposed implementation

1. Add optional committer parallelism and max-parallelism fields to the dynamic sink builder/configuration.
2. Propagate them to the pre-commit/committer transformation through a Flink SinkV2 API capable of configuring the generated committer operator.
3. Keep the current inherited behavior as the default for backward compatibility.
4. Preserve deterministic internal operator UIDs derived from uidPrefix.
5. Add topology tests proving that writer parallelism, committer parallelism and committer max parallelism are independent.
6. Add recovery/rescale tests documenting the keyed-state migration constraints when max parallelism changes.

If Flink SinkV2 cannot currently express this, this issue can track the Iceberg side while the companion Flink issue adds the required framework capability or fixes the autoscaler behavior.

Related Flink issue

Companion Flink autoscaler issue: pending creation on Apache Jira. This issue will be updated with the link.

[pvary]

Could you please check the capabilities of the Flink API, and propose a solution with that in mind?

[Sbaia]

I checked Flink's SinkV2 API in 2.1.2 and current master.

SupportsCommitter only exposes committer creation and serialization. SupportsPreCommitTopology controls the stream before the committer. The actual CommitterOperator is created afterward by Flink's SinkTransformationTranslator, which assigns it the sink transformation's parallelism and max parallelism when unset.

Therefore, Iceberg currently has no public API through which DynamicIcebergSink can configure the generated committer transformation independently.

I propose splitting the change:

1. Flink adds an optional SinkV2 capability exposing committer parallelism and max parallelism.
2. SinkTransformationTranslator applies these values to the generated CommitterOperator.
3. Iceberg adds committerParallelism(...) and committerMaxParallelism(...) to DynamicIcebergSink.Builder.
4. Empty values preserve the current inherited behaviour.

This capability would provide a connector-level bound. The incorrect autoscaler target-rate calculation observed for the multi-input committer remains a separate Flink autoscaler issue.

I will open the companion Flink issue and link it here.