dmvk commented on a change in pull request #18689:
URL: https://github.com/apache/flink/pull/18689#discussion_r803770073
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java
##########
@@ -134,30 +143,53 @@ public JobStatus getJobStatus() {
return JobStatus.RUNNING;
}
+ private void handleFailure(Failure failure) {
+ failureCollection.add(failure);
+ FailureResult failureResult = context.howToHandleFailure(failure);
+ transitionOnFailure(failureResult);
+ }
+
@Override
public void handleGlobalFailure(Throwable cause) {
- handleAnyFailure(cause);
+ handleFailure(Failure.createGlobal(cause));
}
@Override
boolean updateTaskExecutionState(TaskExecutionStateTransition
taskExecutionStateTransition) {
final boolean successfulUpdate =
getExecutionGraph().updateState(taskExecutionStateTransition);
- if (successfulUpdate) {
- if (taskExecutionStateTransition.getExecutionState() ==
ExecutionState.FAILED) {
- Throwable cause =
taskExecutionStateTransition.getError(userCodeClassLoader);
- handleAnyFailure(
- cause == null
- ? new FlinkException(
- "Unknown failure cause. Probably
related to FLINK-21376.")
- : cause);
- }
+ if (successfulUpdate
+ && taskExecutionStateTransition.getExecutionState() ==
ExecutionState.FAILED) {
+ handleFailure(
+ Failure.createLocal(
+ extractError(taskExecutionStateTransition),
+
extractExecutionVertexID(taskExecutionStateTransition)));
}
return successfulUpdate;
}
+ private void transitionOnFailure(FailureResult failureResult) {
Review comment:
nvm, I've missed that `StateWithExecutionGraph` can't always transition
to `restarting` and `failing`
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
