bujjibabukatta opened a new pull request, #17879:
URL: https://github.com/apache/iceberg/pull/17879
## 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.
--
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]
