Sbaia opened a new issue, #17863: URL: https://github.com/apache/iceberg/issues/17863
## 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. - https://github.com/apache/iceberg/issues/17140 - https://github.com/apache/iceberg/pull/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: ```java 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 FLINK_JIRA_URL -- 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]
