docs: 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));
  		}
  	}