class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {

	@Override
	public ClusterClientProvider<ApplicationId> deployApplicationCluster(
			final ClusterSpecification clusterSpecification,
			final ApplicationConfiguration applicationConfiguration) throws ClusterDeploymentException {
		checkNotNull(clusterSpecification);
		checkNotNull(applicationConfiguration);

		final YarnDeploymentTarget deploymentTarget = YarnDeploymentTarget.fromConfig(flinkConfiguration);
		if (YarnDeploymentTarget.APPLICATION != deploymentTarget) {
			throw new ClusterDeploymentException(
					"Couldn't deploy Yarn Application Cluster." +
							" Expected deployment.target=" + YarnDeploymentTarget.APPLICATION.getName() +
							" but actual one was \"" + deploymentTarget.getName() + "\"");
		}

		applicationConfiguration.applyToConfiguration(flinkConfiguration);

		final List<String> pipelineJars = flinkConfiguration.getOptional(PipelineOptions.JARS).orElse(Collections.emptyList());
		Preconditions.checkArgument(pipelineJars.size() == 1, "Should only have one jar");

		try {
			return deployInternal(
					clusterSpecification,
					"Flink Application Cluster",
					YarnApplicationClusterEntryPoint.class.getName(),
					null,
					false);
		} catch (Exception e) {
			throw new ClusterDeploymentException("Couldn't deploy Yarn Application Cluster", e);
		}
	}

	public  YarnClusterDescriptor(
			Configuration flinkConfiguration,
			YarnConfiguration yarnConfiguration,
			YarnClient yarnClient,
			YarnClusterInformationRetriever yarnClusterInformationRetriever,
			boolean sharedYarnClient);

}