Flink: Allow independent parallelism bounds for DynamicIcebergSink committer

[bujjibabukatta]

Summary

Adds committerParallelism(int) and committerMaxParallelism(int) to
DynamicIcebergSink.Builder, letting the pre-commit/committer operator be
scaled independently of writeParallelism(int). Both are optional; when
unset, the committer continues to inherit its parallelism from the upstream
writer topology (unchanged default behavior).

Closes #17863

Root Cause

DynamicIcebergSink implements Flink's SupportsPreCommitTopology, and
builds its own pre-commit/committer operator via
.keyBy(...).transform(...) in addPreCommitTopology(). That
.transform(...) call never set an explicit parallelism, so the operator
silently inherited whatever parallelism the upstream writer chain had.

Since dynamic committers process checkpoint-batched, per-table metadata
rather than per-record data, their scaling needs differ from the writer's.
With no independent bound, autoscalers that estimate demand from edge rates
can scale the committer in lockstep with the writer even though its actual
load is far lower, and the only workarounds (job.autoscaler.vertex.exclude.ids,
a global autoscaler max, or changing job-wide max parallelism) are either
awkward or affect unrelated vertices/keyed-state compatibility.

Fix

• Added FlinkWriteOptions.COMMITTER_PARALLELISM and
  COMMITTER_MAX_PARALLELISM config options.
• Added corresponding accessors to FlinkWriteConf.
• Added DynamicIcebergSink.Builder#committerParallelism(int) and
  #committerMaxParallelism(int), writing into the same writeOptions map
  as the existing writeParallelism(int).
• In addPreCommitTopology(), capture the .transform(...) result and
  conditionally call .setParallelism(...) / .setMaxParallelism(...) on
  it when the new options are set, leaving the operator's .uid(...)
  unchanged so existing checkpoint state remains restorable.

No Flink SinkV2 framework changes were needed — addPreCommitTopology() is
implemented entirely with the standard, public DataStream API, which
already exposes setParallelism/setMaxParallelism on the returned
operator.

[bujjibabukatta]

Hi @sbaia could you please review code and merge the pr?

[bujjibabukatta]

Hi @huaxingao can you please review and merge pull request ?

[Sbaia]

I reviewed the generated SinkV2 topology, and I don't think the current implementation configures the actual committer operator.

The new values are applied to the transformation returned by DynamicIcebergSink.addPreCommitTopology(). That transformation contains DynamicWriteResultAggregator, not DynamicCommitter:

https://github.com/apache/iceberg/pull/17879/files

Flink creates the actual CommitterOperator afterwards, in a separate adjustTransformations call:

https://github.com/apache/flink/blob/release-2.1.2/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java#L259-L294

Because that generated transformation still has default parallelism/max parallelism, Flink assigns it the values from the sink transformation, not from the preceding pre-commit transformation:

https://github.com/apache/flink/blob/release-2.1.2/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java#L425-L440

For example, with:

.writeParallelism(8)
.committerParallelism(1)
.committerMaxParallelism(1)

this patch sets the pre-commit aggregator to parallelism 1, while the generated Sink Committer can remain at parallelism 8. Flink then uses a rebalance edge when upstream and downstream parallelism differ:

https://github.com/apache/flink/blob/release-2.1.2/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java#L901-L920

This is also a correctness concern. SupportsPreCommitTopology requires summaries and committables with the same subtask ID to be processed by the same committer subtask:

https://github.com/apache/flink/blob/release-2.1.2/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/SupportsPreCommitTopology.java#L35-L45

Setting only committerMaxParallelism may appear to work while both operators keep the same parallelism and remain in the same forward group. However, that relies on topology coupling and does not provide independent control of the generated committer. Setting committerParallelism to a different value breaks that assumption.

Could you please add a JobGraph-level regression test that:

1. sets writer parallelism to 8 and committer parallelism/max parallelism to 1;
2. locates both the pre-commit operator and the generated Sink Committer;
3. asserts that the Sink Committer itself has parallelism 1 and max parallelism 1;
4. verifies that the edge preserves the CommittableSummary/committable routing contract?

I expect this test to fail with the current implementation. If so, we still need either a Flink SinkV2 API/translator capability for the generated committer, or a larger Iceberg-side topology redesign. Also, the PR currently changes only the Flink 2.1 module and adds no tests for the new options.