Sbaia commented on PR #17879: URL: https://github.com/apache/iceberg/pull/17879#issuecomment-5476289441
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: ```java .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. -- 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]
