[FLINK-4046] [runtime] Add direct state transition from RESTARTING to FAILED

[tillrohrmann]

A job can get stuck in FAILING if fail is called on a restarting job which has
not yet reset its ExecutionJobVertices, because these vertices would not call
jobVertexInFinalState. This method, however, must be called in order to transition
from FAILING to FAILED. In order to solve the problem, this PR introduces a direct
state transition from RESTARTING to FAILED, if fail is called when being in state
RESTARTING.

[uce]

+1 to merge. Both the direct state transition and test are good.

[uce]

PS: Should the early abort check for the CANCELED state in ExecutionGraph.restart() be extended to any terminal state? The restart method assumes that it's the only one transitioning away from RESTARTING, which is not the case any more. I think it's not needed, but might lead to confusing log messages when fail and restart are called concurrently.

[tillrohrmann]

Thanks for the review @uce. I think you're right and we should also accept the FAILED state when calling restart and simply print a log message instead of throwing the IllegalStateException. But I'm not so sure for any terminal state because we should not have reached the state FINISHED. This usually indicates a failure and should not go unnoticed.

I will add the FAILED check and then merge the PR.

[uce]

Agreed. I had the same thought regarding checking for any final state right after posting this. ;) +1 to merge.