public void suspend ( Throwable suspensionCause ) { assertRunningInJobMasterMainThread ( ) ; if ( state . isTerminalState ( ) ) { // stay in a terminal state return ; } else if ( transitionState ( state , JobStatus . SUSPENDED , suspensionCause ) ) { initFailureCause ( suspensionCause ) ; // make sure no concurrent local actions interfere with the cancellation incrementGlobalModVersion ( ) ; // cancel ongoing scheduling action if ( schedulingFuture != null ) { schedulingFuture . cancel ( false ) ; } final ArrayList < CompletableFuture < Void > > executionJobVertexTerminationFutures = new ArrayList <> ( verticesInCreationOrder . size ( ) ) ; for ( ExecutionJobVertex ejv : verticesInCreationOrder ) { executionJobVertexTerminationFutures . add ( ejv . suspend ( ) ) ; } final ConjunctFuture < Void > jobVerticesTerminationFuture = FutureUtils . waitForAll ( executionJobVertexTerminationFutures ) ; checkState ( jobVerticesTerminationFuture . isDone ( ) , "Suspend needs to happen atomically" ) ; jobVerticesTerminationFuture . whenComplete ( ( Void ignored , Throwable throwable ) -> { if ( throwable != null ) { LOG . debug ( "Could not properly suspend the execution graph." , throwable ) ; } onTerminalState ( state ) ; LOG . info ( "Job {} has been suspended." , getJobID ( ) ) ; } ) ; } else { throw new IllegalStateException ( String . format ( "Could not suspend because transition from %s to %s failed." , state , JobStatus . SUSPENDED ) ) ; } }
Suspends the current ExecutionGraph .