[FLINK-13124] Don't forward exceptions when finishing SourceStreamTask #9090

aljoscha · 2019-07-11T13:46:52Z

What is the purpose of the change

Before, exceptions that occurred after cancelling a source (as the
KafkaConsumer did, for example) would make a job fail when attempting a
"stop-with-savepoint". Now we ignore those exceptions.

Brief change log

add a isFinished flag in SourceStreamTask
check this flag before re-throwing exceptions

Verifying this change

This change added tests and can be verified as follows:

Added unit tests in SourceStreamTaskTest

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

Checkpointing, yes, because it fixes stop-with-savepoint

Documentation

Does this pull request introduce a new feature? no

aljoscha · 2019-07-11T13:47:04Z

cc @knaufk

flinkbot · 2019-07-11T13:48:33Z

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.

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 · 2019-07-11T13:56:49Z

CI report:

6c22974 : FAILURE Build
6b32e06 : SUCCESS Build

knaufk · 2019-07-11T15:35:57Z

stop now works with the fix in my test setup. Thanks @aljoscha.

kl0u

Thanks for the work @aljoscha ! I had some comments that I left on the PR.

kl0u · 2019-07-12T08:45:00Z

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

+	 * want to ignore exceptions thrown after finishing, to ensure shutdown works smoothly.
+	 */
+	private volatile boolean isFinished = false;
+


I would suggest to move this flag to the StreamTask. The reason is that although for now it is used only in the SourceStreamTask, know if the job was normally terminated or cancelled is a property relevant to all the StreamTasks.

kl0u · 2019-07-12T08:50:59Z

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

@@ -147,6 +155,7 @@ protected void cancelTask() {

 	@Override


With the above comments in mind, I would suggest to have a method in the StreamTask like the following:

@VisibleForTesting void finish() throws Exception { finished = true; finishTask(); // this is implemented by subclasses }

(needed in the finishingIgnoresExceptions() test).

And with this, the end of the performCheckpoint should become:

if (isRunning && syncSavepointLatch.isSet()) { final boolean checkpointWasAcked = syncSavepointLatch.blockUntilCheckpointIsAcknowledged(); if (checkpointWasAcked) { finishTask(); } }

kl0u · 2019-07-12T08:51:01Z

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

@@ -118,7 +124,9 @@ protected void performDefaultAction(ActionContext context) throws Exception {
 		}

 		sourceThread.join();
-		sourceThread.checkThrowSourceExecutionException();
+		if (!isFinished) {


Given the above comment, this can be a protected getter method for the finished flag in the StreamTask, available to all StreamTasks.

aljoscha · 2019-07-12T16:57:34Z

Master has moved since I opened this PR, the fix still works, with a change, but I want to investigate a bit further because that change on master removes other new behaviour that my PR was testing in one of the test cases.

aljoscha · 2019-07-13T22:45:41Z

PTAL (please take another look). I now changed this to work with latest master. (there was a change in how the mailbox worked.)

Before, exceptions that occurred after cancelling a source (as the KafkaConsumer did, for example) would make a job fail when attempting a "stop-with-savepoint". Now we ignore those exceptions.

kl0u

@aljoscha Thanks for the work!

Changes look good for the master, but the changes in the mailbox that you refer to, are not present in the release-1.9 branch. I can merge this PR to the master, but for the release-1.9 you should open a new PR, potentially with the solution included in this PR before your last rebasing.

WDYT?

aljoscha · 2019-07-15T10:49:27Z

ok, thanks! could you please merge? (just arrived in china 😅 )

kl0u · 2019-07-15T10:54:29Z

Sure! I will merge now.

kl0u · 2019-07-15T11:04:04Z

Merged on master

1u0 · 2019-07-15T14:49:15Z

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

+		// We tell the mailbox to finish, to prevent any exceptions that might occur during
+		// finishing from leading to a FAILED state. This could happen, for example, when cancelling
+		// sources as part of a "stop-with-savepoint".
+		mailboxProcessor.allActionsCompleted();


Sorry for being late in this PR.
Is there a reason why in master branch version there is mailboxProcessor.allActionsCompleted(); in finishTask() but in release-1.9 branch doesn't have it (#9115)?

On the release-1.9 branch StreamTask.cancel() also doesn't use this method, in fact the MailboxProcessor doesn't yet exist and isn't used like this. However, this solution leads to a race condition, see also #9125.

aljoscha requested a review from kl0u July 11, 2019 13:46

rmetzger added the review=description? label Jul 11, 2019

rmetzger added the component=Connectors/Kafka label Jul 11, 2019

kl0u requested changes Jul 12, 2019

View reviewed changes

kl0u mentioned this pull request Jul 12, 2019

[FLINK-13205][runtime] Make checkpoints injection ordered with stop-with-savepoint #9021

Closed

aljoscha force-pushed the flink13124-stop-with-savepoint-fix branch from 6c22974 to b1330cb Compare July 13, 2019 22:45

[FLINK-13124] Don't forward exceptions when finishing SourceStreamTask

6b32e06

Before, exceptions that occurred after cancelling a source (as the KafkaConsumer did, for example) would make a job fail when attempting a "stop-with-savepoint". Now we ignore those exceptions.

aljoscha force-pushed the flink13124-stop-with-savepoint-fix branch from b1330cb to 6b32e06 Compare July 14, 2019 05:23

kl0u self-requested a review July 15, 2019 10:15

kl0u approved these changes Jul 15, 2019

View reviewed changes

kl0u closed this Jul 15, 2019

1u0 reviewed Jul 15, 2019

View reviewed changes

aljoscha deleted the flink13124-stop-with-savepoint-fix branch July 25, 2019 15:30

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-13124] Don't forward exceptions when finishing SourceStreamTask #9090

[FLINK-13124] Don't forward exceptions when finishing SourceStreamTask #9090

aljoscha commented Jul 11, 2019

aljoscha commented Jul 11, 2019

flinkbot commented Jul 11, 2019

flinkbot commented Jul 11, 2019 •

edited

knaufk commented Jul 11, 2019

kl0u left a comment

kl0u Jul 12, 2019

kl0u Jul 12, 2019

kl0u Jul 12, 2019

aljoscha commented Jul 12, 2019

aljoscha commented Jul 13, 2019

kl0u left a comment

aljoscha commented Jul 15, 2019

kl0u commented Jul 15, 2019

kl0u commented Jul 15, 2019

1u0 Jul 15, 2019

aljoscha Jul 16, 2019

[FLINK-13124] Don't forward exceptions when finishing SourceStreamTask #9090

[FLINK-13124] Don't forward exceptions when finishing SourceStreamTask #9090

Conversation

aljoscha commented Jul 11, 2019

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

aljoscha commented Jul 11, 2019

flinkbot commented Jul 11, 2019

Review Progress

flinkbot commented Jul 11, 2019 • edited

CI report:

knaufk commented Jul 11, 2019

kl0u left a comment

Choose a reason for hiding this comment

kl0u Jul 12, 2019

Choose a reason for hiding this comment

kl0u Jul 12, 2019

Choose a reason for hiding this comment

kl0u Jul 12, 2019

Choose a reason for hiding this comment

aljoscha commented Jul 12, 2019

aljoscha commented Jul 13, 2019

kl0u left a comment

Choose a reason for hiding this comment

aljoscha commented Jul 15, 2019

kl0u commented Jul 15, 2019

kl0u commented Jul 15, 2019

1u0 Jul 15, 2019

Choose a reason for hiding this comment

aljoscha Jul 16, 2019

Choose a reason for hiding this comment

flinkbot commented Jul 11, 2019 •

edited