class KubernetesResourceManagerDriver extends AbstractResourceManagerDriver<KubernetesWorkerNode> {

	@Override
	public CompletableFuture<KubernetesWorkerNode> requestResource(TaskExecutorProcessSpec taskExecutorProcessSpec) {
		final KubernetesTaskManagerParameters parameters =
				createKubernetesTaskManagerParameters(taskExecutorProcessSpec);
		final KubernetesPod taskManagerPod =
				KubernetesTaskManagerFactory.buildTaskManagerKubernetesPod(parameters);
		final String podName = taskManagerPod.getName();
		final CompletableFuture<KubernetesWorkerNode> requestResourceFuture = new CompletableFuture<>();

		requestResourceFutures.put(podName, requestResourceFuture);

		log.info("Creating new TaskManager pod with name {} and resource <{},{}>.",
				podName,
				parameters.getTaskManagerMemoryMB(),
				parameters.getTaskManagerCPU());

		// When K8s API Server is temporary unavailable, `kubeClient.createTaskManagerPod` might fail immediately.
		// In case of pod creation failures, we should wait for an interval before trying to create new pods.
		// Otherwise, ActiveResourceManager will always re-requesting the worker, which keeps the main thread busy.
		final CompletableFuture<Void> createPodFuture =
				podCreationCoolDown.thenCompose((ignore) -> kubeClient.createTaskManagerPod(taskManagerPod));

		FutureUtils.assertNoException(
				createPodFuture.handleAsync((ignore, exception) -> {
					if (exception != null) {
						log.warn("Could not create pod {}, exception: {}", podName, exception);
						tryResetPodCreationCoolDown();
						CompletableFuture<KubernetesWorkerNode> future =
								requestResourceFutures.remove(taskManagerPod.getName());
						if (future != null) {
							future.completeExceptionally(exception);
						}
					} else {
						log.info("Pod {} is created.", podName);
					}
					return null;
				}, getMainThreadExecutor()));

		return requestResourceFuture;
	}

	public  KubernetesResourceManagerDriver(
			Configuration flinkConfig,
			FlinkKubeClient kubeClient,
			KubernetesResourceManagerDriverConfiguration configuration);

}