class PerJobMiniClusterFactory {

	public CompletableFuture<JobClient> submitJob(JobGraph jobGraph, ClassLoader userCodeClassloader) throws Exception {
		MiniClusterConfiguration miniClusterConfig = getMiniClusterConfig(jobGraph.getMaximumParallelism());
		MiniCluster miniCluster = miniClusterFactory.apply(miniClusterConfig);
		miniCluster.start();

		return miniCluster
			.submitJob(jobGraph)
			.thenApplyAsync(FunctionUtils.uncheckedFunction(submissionResult -> {
				org.apache.flink.client.ClientUtils.waitUntilJobInitializationFinished(
					() -> miniCluster.getJobStatus(submissionResult.getJobID()).get(),
					() -> miniCluster.requestJobResult(submissionResult.getJobID()).get(),
					userCodeClassloader);
				return submissionResult;
			}))
			.thenApply(result -> new MiniClusterJobClient(
					result.getJobID(),
					miniCluster,
					userCodeClassloader,
					MiniClusterJobClient.JobFinalizationBehavior.SHUTDOWN_CLUSTER))
			.whenComplete((ignored, throwable) -> {
				if (throwable != null) {
					// We failed to create the JobClient and must shutdown to ensure cleanup.
					shutDownCluster(miniCluster);
				}
			})
			.thenApply(Function.identity());
	}

	private  PerJobMiniClusterFactory(
			Configuration configuration,
			Function<? super MiniClusterConfiguration, ? extends MiniCluster> miniClusterFactory);

}