[FLINK-11250][runtime] Added method init for RecordWriter for initialization resources(OutputFlusher) outside of constructor

[akalash]

What is the purpose of the change

This PR resolves the problem with closing the OutputFlusher when an exception happens before the task restore.

Brief change log

• New method init for RecordWriter created
• Moved the INITIALIZING transition under cleanUp try-catch block

Verifying this change

This change added tests and can be verified as follows:

• *Added tests for initialization of RecordWriter *
• Added test for check closing of OutputFlusher from Task

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]

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 d7bfc75 (Tue Sep 07 17:10:00 UTC 2021)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{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.

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]

CI report:

• 9295f1f Azure: FAILURE

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

[curcur]

Hey @akalash , thanks for the fix. LGTM mostly.
but shouldn't [refactor][runtime] Moved the INITIALIZING transition under cleanUp try-catch block be enough to fix the problem?

Why the init method has to be pulled out and called explicitly here restoreInternal()?

[curcur]

why called here separately?

What's the problem of starting the thread in constructor?

[akalash]

We somehow should guarantee that all resources will be closed in case of an error. Before my changes, if the StreamTask fails with the error right after recordWriter would be created, nobody closes it which leads to the leak. So semantic should be as following:

allocateResources;
try {
 // do something
} catch(Throwable ex) {
  releaseResources;
}

But it is difficult to follow this rule with the current implementation. For example, look at StreamTask#createRecordWriters, if creating the second record writer fails we lost the link to the first record writer, so it would be impossible to close it.

So in my opinion, we have two choices here: to have the init method which would be invoked under try-catch block(it is exactly what I did). or rewrite a code in such a way that exception in any constructor(ex. StreamTask) guarantee releasing the earlier allocated resources in this constructor.

[akalash]

shouldn't [refactor][runtime] Moved the INITIALIZING transition under cleanUp try-catch block be enough to fix the problem?

Yes, you are almost right. This commit resolves the problem related to the exception after StreamTask was created but this commit doesn't resolve the problem if the exception happens inside of StreamTask constructor right after recordWriter is created. You can check my test testRecordWriterClosedOnStreamTaskConstructorError which reproduce this problem.

[curcur]

I think there are other resources that might be null as well. Not just
channelIOExecutor and mailboxProcessor

Let's take releaseOutputResources for example,
both operatorChain and recordWriter can possibly be null if thrown exception from the constructor?

[akalash]

Yes, of course, there are more possible NPE but it is not problem since all of that NPE could happen only inside runAndSuppressThrowable which helps to handle it correctly. at least, all variables will checked

[curcur]

er... I think runAndSuppressThrowable is not designed for this purpose, but coincidentally yes ^-^.

[curcur]

Hey @akalash , thanks for updating the PR.

Please take a look at my comments. Besides of this, I think this try-catch solution + adding non-null check is not a clean way of solving the problem. In the end, the solution may as complicated as extract all resource initialization outside of the constructor (you have to check each of the member variables whether they are null or not).

I think the right way is to extract the resource initialization out of streamtask. But I do understand that may need careful walk through the code and clean-up. So I am also fine to take this bug fix as two steps:

1. Fix the task status transition -> initialization resource leak
2. Fix the leak in the stream task constructor.

[akalash]

@curcur, Do I understand correctly that you propose to do nothing in the constructor in this PR but leave only changes related to transition state?

[akalash]

@curcur , I left in PR only changes about transition state. And I also have created the extra ticket - https://issues.apache.org/jira/browse/FLINK-24294

[curcur]

Thanks @akalash for fixing
The failure does not seem to be related.

merged.