[FLINK-22972][datastream] Remove StreamOperator#dispose in favour of close and finish #16351

dawidwys · 2021-07-02T07:15:26Z

What is the purpose of the change

This PR cleans up StreamOperator API in regards to the termination phase and introduces a clean finish() method for flushing all records without releasing resources.

The StreamOperator#close method which is supposed to flush all records, but at the same time, currently, it closes all resources, including connections to external systems. We need separate methods for flushing and closing resources because we might need the connections when performing the final checkpoint, once all records are flushed. Moreover, the logic for closing resources is duplicated in the StreamOperator#dispose method.

Verifying this change

All existing tests pass.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2021-07-02T07:19:21Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit c134bf9 (Thu Sep 23 17:55:59 UTC 2021)

✅no warnings

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-07-02T07:41:29Z

CI report:

e6e93af UNKNOWN
80062e9 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

dawidwys · 2021-07-02T12:59:59Z

@flinkbot run azure

gaoyunhaii

Hi @dawidwys very thanks for opening the PR! The PR looks in general very good to me and I only left few comments.

Besides the comments inline,

it seems we might also need to modify the document task_lifecycle.md
It seems there are also some operators: PythonTimestampsAndWatermarksOperator, CollectSinkOperator, AbstractMapBundleOperator, StreamSortOperator, RowTimeMiniBatchAssginerOperator, TestBoundedMultipleInputOperator and TimestampITCase.CustomOperator, some of their logic in close() should also need to be migrated to be in finish() and some of them would also emit records in close()

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java

gaoyunhaii · 2021-07-05T08:23:49Z

...ming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java

-     */
    @Override
-    public void dispose() throws Exception {
+    public void close() throws Exception {


Although it is definitely not introduced in this PR, I have a bit concern here in that we rely on the subclass to call super.close(), otherwise the statehandler won't get cleaned up, which might further cause resource leak (like the memory occupied by rocksdb). Currently it seems we have operators like GenericWriteAheadSink , TemporalProcessTimeJoinOperator, TemporalProcessTimeJoinOperator and WatermarkAssignerOperator that indeed do not call super.close(). Perhaps we could introduce a final method closeAndCleanupState that get called by the framework, and call close() in that method~?

That's definitely a good observation. I am a bit hesitant to introducing the change as it implies changing again the API of the StreamOperator. Underneath the framework calls StreamOperator#close. If we wanted to introduce closeAndCleanupState the way you described it that it'd call close(), we'd need to do it in the StreamOperator. If we wanted to do it the other way around and make AbstractStreamOperator#close final and call abstract closeAndCleanupState or similar we'd need to change all operators and most probably all user's operators as it's virtually impossible to implement an operator without extending one of the AbstractStreamOperatorV2.

How about we create a JIRA ticket for that and we try to fix it once we work making the operator API "more public". There is a desire to expose e.g. the MailboxProcessor and similar features in a better thought through manner.

Very thanks for the explanation! It indeed makes a lot of sense to me and I also agree with the plan.

And for now perhaps we first complement super.close() for those three operators? Since some of them indeed used states~

Ah, right. Missed that bit 🤦 Yes, will update those three operators.

gaoyunhaii · 2021-07-05T08:26:04Z

...ng-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java

-     */
    @Override
-    public void dispose() throws Exception {
+    public void close() throws Exception {


And here there might be a similar concern as the AbstractStreamOperator.

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java

...-state-processing-api/src/main/java/org/apache/flink/state/api/output/BoundedStreamTask.java

.../org/apache/flink/table/runtime/operators/multipleinput/MultipleInputStreamOperatorBase.java

…texts

…close and finish This commit cleans up StreamOperator API in regards to the termination phase and introduces a clean finish() method for flushing all records without releasing resources. The StreamOperator#close method which is supposed to flush all records, but at the same time, currently, it closes all resources, including connections to external systems. We need separate methods for flushing and closing resources because we might need the connections when performing the final checkpoint, once all records are flushed. Moreover, the logic for closing resources is duplicated in the StreamOperator#dispose method.

gaoyunhaii · 2021-07-06T02:41:45Z

Very thanks @dawidwys for the updates! The PR LGTM now~

rmetzger added the review=description? label Jul 2, 2021

rmetzger added component=API/Core component=Runtime/Task labels Jul 2, 2021

dawidwys force-pushed the flink-22972 branch from b50852d to 68484a6 Compare July 5, 2021 07:11

gaoyunhaii reviewed Jul 5, 2021

View reviewed changes

dawidwys added 8 commits July 5, 2021 12:34

[FLINK-22972][datastream] Forbid emitting from closed StreamSourceCon…

7ef3908

…texts

Update wrong javadocs

feecb1d

Call endInput before finish

10f34dd

Fix operators that called close instead of finish

8a2d862

Rename DisposeException to CloseException

60b9f22

Update task_lifecycle.md

facdbf3

Remove unused TestBoundedMultipleInputOperator

0567347

dawidwys force-pushed the flink-22972 branch from 80062e9 to 0567347 Compare July 5, 2021 10:36

Call super.close()

c134bf9

dawidwys closed this in 541f430 Jul 6, 2021

[FLINK-22972][datastream] Remove StreamOperator#dispose in favour of close and finish #16351

[FLINK-22972][datastream] Remove StreamOperator#dispose in favour of close and finish #16351

Uh oh!

Conversation

dawidwys commented Jul 2, 2021

What is the purpose of the change

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Jul 2, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

flinkbot commented Jul 2, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

dawidwys commented Jul 2, 2021

Uh oh!

gaoyunhaii left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gaoyunhaii Jul 5, 2021

Choose a reason for hiding this comment

Uh oh!

dawidwys Jul 5, 2021

Choose a reason for hiding this comment

Uh oh!

gaoyunhaii Jul 5, 2021

Choose a reason for hiding this comment

Uh oh!

dawidwys Jul 5, 2021

Choose a reason for hiding this comment

Uh oh!

gaoyunhaii Jul 5, 2021

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gaoyunhaii commented Jul 6, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Jul 2, 2021 •

edited

Loading

flinkbot commented Jul 2, 2021 •

edited

Loading