Runs a continuation when a task completes successfully, forwarding along
{@link java.lang.Exception}s or cancellation.
511214
public <TContinuationResult> Task<TContinuationResult> onSuccess(
      final Continuation<TResult, TContinuationResult> continuation) {
    return onSuccess(continuation, IMMEDIATE_EXECUTOR, null);
  }
878203
@Override
	public CompletableFuture<Void> onStop() {
		log.info("Stopping TaskExecutor {}.", getAddress());

		Throwable throwable = null;

		if (resourceManagerConnection != null) {
			resourceManagerConnection.close();
		}

		for (JobManagerConnection jobManagerConnection : jobManagerConnections.values()) {
			try {
				disassociateFromJobManager(jobManagerConnection, new FlinkException("The TaskExecutor is shutting down."));
			} catch (Throwable t) {
				throwable = ExceptionUtils.firstOrSuppressed(t, throwable);
			}
		}

		jobManagerHeartbeatManager.stop();

		resourceManagerHeartbeatManager.stop();

		try {
			stopTaskExecutorServices();
		} catch (Exception e) {
			throwable = ExceptionUtils.firstOrSuppressed(e, throwable);
		}

		if (throwable != null) {
			return FutureUtils.completedExceptionally(new FlinkException("Error while shutting the TaskExecutor down.", throwable));
		} else {
			log.info("Stopped TaskExecutor {}.", getAddress());
			return CompletableFuture.completedFuture(null);
		}
	}
390388
@Override
    public void stop(Throwable cause) throws Exception {
        stopping = true;
        if (task != null) {
            task.cancel(true);
        }
        super.stop(cause);
    }
1122015
public synchronized boolean fail(ServiceException ex) {
        if (completed) {
            return false;
        }
        this.completed = true;
        this.ex = ex;
        notifyAll();
        return true;
    }
1055515
static ContinuableFuture<Void> execute(final Try.Runnable<? extends Exception> action) {
        return execute(new FutureTask<Void>(FN.toCallable(action)), SERIAL_EXECUTOR);
    }
607334
@Override
    public void waitForFinish() throws IOException
    {
        if (future == null)
        {
            throw new IllegalStateException("not started");
        }
        try
        {
            future.get();
        }
        catch (InterruptedException | ExecutionException ex)
        {
            throw new IOException(ex);
        }
    }
942035
protected <T> Future<T> invokeSubmit(ExecutorService executor, Callable<T> task) {
    return executor.submit(wrapTask(task));
  }
360469
@VisibleForTesting
    protected void executionComplete(Throwable exception) {
        Collection<QueueItem> toFail = null;
        Throwable failEx = null;
        synchronized (this.stateLock) {
            // Release the spot occupied by this item's execution.
            this.activeCount--;
            if (exception != null && !this.closed) {
                // Need to fail all future items and close to prevent new items from being processed.
                failEx = new ProcessingException("A previous item failed to commit. Cannot process new items.", exception);
                toFail = new ArrayList<>(this.pendingItems);
                this.pendingItems.clear();
                this.closed = true;
            }

            if (this.emptyNotifier != null && this.activeCount == 0 && this.pendingItems.isEmpty()) {
                // We were asked to notify when we were empty.
                this.emptyNotifier.release();
                this.emptyNotifier = null;
            }
        }

        if (toFail != null) {
            for (QueueItem q : toFail) {
                q.result.completeExceptionally(failEx);
            }

            return;
        }

        // We need to ensure the items are still executed in order. Once out of the main sync block, it is possible that
        // this callback may be invoked concurrently after various completions. As such, we need a special handler for
        // these, using its own synchronization, different from the main one, so that we don't hold that stateLock for too long.
        synchronized (this.processingLock) {
            while (true) {
                QueueItem toProcess;
                synchronized (this.stateLock) {
                    if (hasCapacity() && !this.pendingItems.isEmpty()) {
                        // We have spare capacity and we have something to process. Dequeue it and reserve the spot.
                        toProcess = this.pendingItems.pollFirst();
                        this.activeCount++;
                    } else {
                        // No capacity left or no more pending items.
                        break;
                    }
                }

                Futures.completeAfter(() -> processInternal(toProcess.data), toProcess.result);
            }
        }
    }
866018
public <TContinuationResult> Task<TContinuationResult> onSuccessTask(
      final Continuation<TResult, Task<TContinuationResult>> continuation) {
    return onSuccessTask(continuation, IMMEDIATE_EXECUTOR);
  }
1051919
@Override
	public final void run () {
		Status result = execute();
		if (result == null) throw new IllegalStateException("Invalid status 'null' returned by the execute method");
		switch (result) {
		case SUCCEEDED:
			success();
			return;
		case FAILED:
			fail();
			return;
		case RUNNING:
			running();
			return;
		default:
			throw new IllegalStateException("Invalid status '" + result.name() + "' returned by the execute method");
		}
	}
953252
@Override
    public void onSuccess(final DMRResponse response) {
        ModelNode result = response.get();
        if (!result.hasDefined(OUTCOME) || result.isFailure()) {
            onFailedOutcome(result);
            abort();
        } else {
            onSuccess(result);
            proceed();
        }
    }
945258
public Task<Void> continueWhile(Callable<Boolean> predicate,
      Continuation<Void, Task<Void>> continuation, CancellationToken ct) {
    return continueWhile(predicate, continuation, IMMEDIATE_EXECUTOR, ct);
  }
242077
@FFDCIgnore(Throwable.class)
    void run() {
        final boolean trace = TraceComponent.isAnyTracingEnabled();

        if (!state.setRunning()) {
            if (trace && tc.isDebugEnabled())
                Tr.debug(this, tc, "unable to run", state.get());
            nsRunEnd = System.nanoTime();
            if (callback != null)
                callback.onEnd(task, this, null, true, 0, null); // aborted, queued task will never run
            return;
        }

        boolean aborted = true;
        Object callbackContext = null;
        thread = Thread.currentThread();
        try {
            if (callback == null)
                aborted = false;
            else {
                callbackContext = callback.onStart(task, this);
                aborted = state.get() == CANCELED;
            }

            if (aborted)
                nsRunEnd = System.nanoTime();
            else {
                T t;
                if (callable == null) {
                    runnable.run();
                    t = predefinedResult;
                } else {
                    t = callable.call();
                }

                nsRunEnd = System.nanoTime();

                if (result.compareAndSet(state, t)) {
                    state.releaseShared(SUCCESSFUL);
                    if (latch != null)
                        latch.countDown(t);
                } else if (Integer.valueOf(CANCELED).equals(result.get())) {
                    if (trace && tc.isDebugEnabled())
                        Tr.debug(this, tc, "canceled during/after run");
                    // Prevent dirty read of state during onEnd before the canceling thread transitions the state to CANCELING/CANCELED
                    while (state.get() == RUNNING)
                        Thread.yield();
                }

                if (trace && tc.isDebugEnabled())
                    Tr.debug(this, tc, "run", t);
            }

            if (callback != null)
                try {
                    callback.onEnd(task, this, callbackContext, aborted, 0, null);
                } catch (Throwable x) {
                }
        } catch (Throwable x) {
            if (trace && tc.isDebugEnabled())
                Tr.debug(this, tc, "run", x);

            nsRunEnd = System.nanoTime();

            if (result.compareAndSet(state, x)) {
                state.releaseShared(aborted ? ABORTED : FAILED);
                if (latch != null)
                    latch.countDown();
            }

            if (callback != null)
                callback.onEnd(task, this, callbackContext, aborted, 0, x);
        } finally {
            thread = null;
            // Prevent accidental interrupt of subsequent operations by awaiting the transition from CANCELING to CANCELED
            while (state.get() == CANCELING)
                Thread.yield();
        }
    }
1260276
@Override
	public void onTaskFailure(Execution taskExecution, Throwable cause) {

		executionGraph.getJobMasterMainThreadExecutor().assertRunningInMainThread();

		// to better handle the lack of resources (potentially by a scale-in), we
		// make failures due to missing resources global failures 
		if (cause instanceof NoResourceAvailableException) {
			LOG.info("Not enough resources to schedule {} - triggering full recovery.", taskExecution);
			executionGraph.failGlobal(cause);
			return;
		}

		LOG.info("Recovering task failure for {} (#{}) via individual restart.", 
				taskExecution.getVertex().getTaskNameWithSubtaskIndex(), taskExecution.getAttemptNumber());

		numTaskFailures.inc();

		// trigger the restart once the task has reached its terminal state
		// Note: currently all tasks passed here are already in their terminal state,
		//       so we could actually avoid the future. We use it anyways because it is cheap and
		//       it helps to support better testing
		final CompletableFuture<ExecutionState> terminationFuture = taskExecution.getTerminalStateFuture();
		terminationFuture.thenRun(
			() -> performExecutionVertexRestart(taskExecution.getVertex(), taskExecution.getGlobalModVersion()));
	}
503159
private boolean waitForTasksToFinish() throws Exception {
        
        Boolean succeeded = true;
        
        // Tracking state for multiple exceptions, if any
        List<StackTraceElement> stackTraces = new ArrayList<StackTraceElement>();
        StringBuilder sb = new StringBuilder();
        int exCount = 0;
        
        for (Future<Boolean> future : futures) {
            try {
                if (!future.get()) {
                    cancel();
                    succeeded = false;
                }
            }
            catch (Exception e) {
                error.compareAndSet(null, e);
                cancel();
                succeeded = false;
                
                exCount++;
                sb.append("ex" + exCount + ": ").append(e.getMessage()).append("\n");
                StackTraceElement[] stackTrace = e.getStackTrace();
                if (stackTrace != null && stackTrace.length > 0) {
                    StackTraceElement delimiterSE = new StackTraceElement("StackTrace: ex" + exCount, "", "", 0);
                    stackTraces.add(delimiterSE);
                    for (StackTraceElement se : stackTrace) {
                        stackTraces.add(se);
                    }
                }
            }
        }
        
        if (this.rowFunction instanceof Flushable) {
            ((Flushable)rowFunction).flush();
        }
        
        if (exCount > 0) {
            String exMessage = sb.toString();
            StackTraceElement[] seArray = stackTraces.toArray(new StackTraceElement[stackTraces.size()]);
            Exception ex = new Exception(exMessage);
            ex.setStackTrace(seArray);
            throw ex;
        }
        return succeeded;
    }
497856
public <TContinuationResult> Task<TContinuationResult> onSuccessTask(
      final Continuation<TResult, Task<TContinuationResult>> continuation,
      CancellationToken ct) {
    return onSuccessTask(continuation, IMMEDIATE_EXECUTOR, ct);
  }
1072108
public <TContinuationResult> Task<TContinuationResult> onSuccessTask(
      final Continuation<TResult, Task<TContinuationResult>> continuation, Executor executor,
      final CancellationToken ct) {
    return continueWithTask(new Continuation<TResult, Task<TContinuationResult>>() {
      @Override
      public Task<TContinuationResult> then(Task<TResult> task) {
        if (ct != null && ct.isCancellationRequested()) {
          return Task.cancelled();
        }

        if (task.isFaulted()) {
          return Task.forError(task.getError());
        } else if (task.isCancelled()) {
          return Task.cancelled();
        } else {
          return task.continueWithTask(continuation);
        }
      }
    }, executor);
  }