pvary opened a new pull request, #23876: URL: https://github.com/apache/flink/pull/23876
## What is the purpose of the change This is the a revisited non-breaking implementation of [FLIP-371: Provide initialization context for Committer creation in TwoPhaseCommittingSink](https://cwiki.apache.org/confluence/display/FLINK/FLIP-371%3A+Provide+initialization+context+for+Committer+creation+in+TwoPhaseCommittingSink). Reverts the previous version [c3a07f98e5d1d7624adc967932f57d31355d9ddd](https://github.com/apache/flink/commit/c3a07f98e5d1d7624adc967932f57d31355d9ddd) with [c7625d5fa62a6e9a182f39f53fb7e5626105f3b0](https://github.com/apache/flink/commit/c7625d5fa62a6e9a182f39f53fb7e5626105f3b0). Then adds the new implementation [99ec936966af527598ca49712c1263bc4aa03c15](https://github.com/apache/flink/commit/99ec936966af527598ca49712c1263bc4aa03c15) based on the [discussion](https://lists.apache.org/thread/h6nkgth838dlh5s90sd95zd6hlsxwx57) on the mailing list. ## Brief change log API changes: - `TwoPhaseCommittingSink` - new `createCommitter` method with default implementation and deprecation - `CommitterInitContext` - new object for committer initialization and metrics - `SinkCommitterMetricGroup` - new metrics group for committer related metrics - `Sink.createWriter` - parameter changed to `WriterInitContext` with default implementation and deprecation - `StatefulSink.createWriter` and `TwoPhaseCommittingSink.createWriter` method declarations are removed - and they fall back to the `Sink.createWriter`, which is backward compatible, since the return type is wider Implementation: - `InitContext` - new common context base class for `CommitterInitContext` and `WriterInitContext` - `WriterInitContext` - replacement for `Sink.InitContext` - `CommitterInitContext` - new context for committer creation - `InitContextBase` - the common base for the `WriterInitContext` and `CommitterInitContext` implementation - `Sink.createWriter(InitContext context)` is deprecated in favour of `createWriter(WriterInitContext context)` with default implementation - `Sink.createWriter(WriterInitContext context)` is created with default implementation which wraps the `WriterInitContext` with `InitContextWrapper` to the old `Sink.InitContext` and calls the original method - `StatefulSink.createWriter` and the `TwoPhaseCommitingSink.createWriter` methods are removed - they fall back to the `Sink.createWriter` method which is backward compatible - `SinkWriterOperator` - to follow the `WriterInitContext` related changes - `StatefulSinkWriterStateHandler` - Here we check if the writer created by `createWriter` is a `StatefulSinkWriter` - `InternalSinkCommitterMetricGroup` - implementation for `SinkCommitterMetricGroup` - `CommitterOperator` - to create the `CommitterInitContext` and `SinkCommitterMetricGroup`. And modify the initialization to use the new `Committer` creation method, and create the metrics group - Updating the metrics in the `CommitRequestImpl` and `CheckpointCommittableManagerImpl` - Propagate the new metricGroup object to finally arrive to the `CommitRequestImpl` where we need it (`CommiterOperator` -> `CommittableCollector` -> `CheckpointCommittableManagerImpl` -> `SubtaskCommittableManager` -> `SubtaskCommittableManager` -> `CommitRequestImpl`) - SinkAdapterV1 changes to use the new Committer and writer creation method, and metrics group (`SinkAdapterV1`, `GlobalCommitterSerializer`, `GlobalCommitterOperator`) - Updating the tests to propagate the metric group - Adding a new test case to `SinkV2MetricsITCase` to make sure that the metrics are propagated as expected. ## Verifying this change Added a new unit test, and modified the old ones to use the new method. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes - The serializers: no - The runtime per-record code paths (performance sensitive): no - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? yes - If yes, how is the feature documented? not documented - might be worth to check the documentation of SinkV2 if it is exists, to update it. -- 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]
