[FLINK-8900] [yarn] Set correct application status when job is finished #5944
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Brings the logging definition in sync with other projects. Updates the classname for the suppressed logger in Netty to account for the new shading model introduced in Flink 1.4.
|
The test failure is an unrelated test flakeyness . |
|
I will try it out. |
GJL
reviewed
May 1, 2018
| @@ -109,7 +119,11 @@ public MiniDispatcher( | |||
|
|
|||
| if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) { | |||
| // terminate the MiniDispatcher once we served the first JobResult successfully | |||
| jobResultFuture.whenComplete((JobResult ignored, Throwable throwable) -> shutDown()); | |||
| jobResultFuture.whenComplete((JobResult result, Throwable throwable) -> { | |||
There was a problem hiding this comment.
throwable isn't used. If jobResultFuture cannot be completed exceptionally, thenAccept should be used.
| @@ -131,6 +133,17 @@ protected JobGraph retrieveJobGraph(Configuration configuration) throws FlinkExc | |||
| } | |||
| } | |||
|
|
|||
| @Override | |||
| protected void registerShutdownActions(CompletableFuture<ApplicationStatus> terminationFuture) { | |||
| terminationFuture.whenComplete((status, throwable) -> | |||
There was a problem hiding this comment.
throwable isn't used. If terminationFuture cannot be completed exceptionally, thenAccept should be used.
| jobResultFuture.whenComplete((JobResult result, Throwable throwable) -> { | ||
| ApplicationStatus status = result.getSerializedThrowable().isPresent() ? | ||
| ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED; | ||
| jobTerminationFuture.complete(status); |
There was a problem hiding this comment.
nit:
I think the functional way would be:
jobTerminationFuture.complete(result.getSerializedThrowable()
.map(serializedThrowable -> ApplicationStatus.FAILED)
.orElse(ApplicationStatus.SUCCEEDED));
There was a problem hiding this comment.
True, I had it like that initially, but found the above version more readable in the end, because we don't really use the serializedThrowable (making the map() a bit strange).
StephanEwen
added a commit
to StephanEwen/flink
that referenced
this pull request
May 3, 2018
StephanEwen
added a commit
to StephanEwen/flink
that referenced
this pull request
May 3, 2018
sampathBhat
pushed a commit
to sampathBhat/flink
that referenced
this pull request
Jul 26, 2018
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
When finite Flink applications (batch jobs) are sent to YARN in the detached mode, the final status is currently always the same, because the job's result is not passed to the logic that initiates the application shutdown.
This PR forwards the final job status via a future that is used to register the shutdown handlers.
Brief change log
JobTerminationFuturein theMiniDispatcherVerifying this change
Run the batch job as described above on YARN to succeed, check that the final application status is successful.
Run the batch job with a parameter to a non existing input file on YARN, check that the final application status is failed.
Does this pull request potentially affect one of the following parts:
@Public(Evolving): (yes / no)Documentation