public void failGlobal ( Throwable t ) { assertRunningInJobMasterMainThread ( ) ; while ( true ) { JobStatus current = state ; // stay in these states if ( current == JobStatus . FAILING || current == JobStatus . SUSPENDED || current . isGloballyTerminalState ( ) ) { return ; } else if ( transitionState ( current , JobStatus . FAILING , t ) ) { initFailureCause ( t ) ; // make sure no concurrent local or global actions interfere with the failover final long globalVersionForRestart = incrementGlobalModVersion ( ) ; final CompletableFuture < Void > ongoingSchedulingFuture = schedulingFuture ; // cancel ongoing scheduling action if ( ongoingSchedulingFuture != null ) { ongoingSchedulingFuture . cancel ( false ) ; } // we build a future that is complete once all vertices have reached a terminal state final ArrayList < CompletableFuture < ? > > futures = new ArrayList <> ( verticesInCreationOrder . size ( ) ) ; // cancel all tasks (that still need cancelling) for ( ExecutionJobVertex ejv : verticesInCreationOrder ) { futures . add ( ejv . cancelWithFuture ( ) ) ; } final ConjunctFuture < Void > allTerminal = FutureUtils . waitForAll ( futures ) ; allTerminal . whenComplete ( ( Void ignored , Throwable throwable ) -> { if ( throwable != null ) { transitionState ( JobStatus . FAILING , JobStatus . FAILED , new FlinkException ( "Could not cancel all execution job vertices properly." , throwable ) ) ; } else { allVerticesInTerminalState ( globalVersionForRestart ) ; } } ) ; return ; } // else: concurrent change to execution state, retry } }
Fails the execution graph globally . This failure will not be recovered by a specific failover strategy but results in a full restart of all tasks .