class ApplicationDispatcherBootstrap extends AbstractDispatcherBootstrap {

	@VisibleForTesting
	CompletableFuture<Acknowledge> runApplicationAndShutdownClusterAsync(
			final DispatcherGateway dispatcher,
			final ScheduledExecutor scheduledExecutor) {

		applicationCompletionFuture = fixJobIdAndRunApplicationAsync(dispatcher, scheduledExecutor);

		return applicationCompletionFuture
				.handle((r, t) -> {
					final ApplicationStatus applicationStatus;
					if (t != null) {

						final Optional<JobCancellationException> cancellationException =
								ExceptionUtils.findThrowable(t, JobCancellationException.class);

						if (cancellationException.isPresent()) {
							// this means the Flink Job was cancelled
							applicationStatus = ApplicationStatus.CANCELED;
						} else if (t instanceof CancellationException) {
							// this means that the future was cancelled
							applicationStatus = ApplicationStatus.UNKNOWN;
						} else {
							applicationStatus = ApplicationStatus.FAILED;
						}

						LOG.warn("Application {}: ", applicationStatus, t);
					} else {
						applicationStatus = ApplicationStatus.SUCCEEDED;
						LOG.info("Application completed SUCCESSFULLY");
					}
					return dispatcher.shutDownCluster(applicationStatus);
				})
				.thenCompose(Function.identity());
	}

	public  ApplicationDispatcherBootstrap(
			final PackagedProgram application,
			final Collection<JobGraph> recoveredJobs,
			final Configuration configuration);

}