class HadoopConfMountDecorator extends AbstractKubernetesStepDecorator {

	@Override
	public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
		Volume hadoopConfVolume;

		final Optional<String> existingConfigMap = kubernetesParameters.getExistingHadoopConfigurationConfigMap();
		if (existingConfigMap.isPresent()) {
			hadoopConfVolume = new VolumeBuilder()
				.withName(Constants.HADOOP_CONF_VOLUME)
				.withNewConfigMap()
					.withName(existingConfigMap.get())
					.endConfigMap()
				.build();
		} else {
			final Optional<String> localHadoopConfigurationDirectory = kubernetesParameters.getLocalHadoopConfigurationDirectory();
			if (!localHadoopConfigurationDirectory.isPresent()) {
				return flinkPod;
			}

			final List<File> hadoopConfigurationFileItems = getHadoopConfigurationFileItems(localHadoopConfigurationDirectory.get());
			if (hadoopConfigurationFileItems.isEmpty()) {
				LOG.warn("Found 0 files in directory {}, skip to mount the Hadoop Configuration ConfigMap.",
					localHadoopConfigurationDirectory.get());
				return flinkPod;
			}

			final List<KeyToPath> keyToPaths = hadoopConfigurationFileItems.stream()
				.map(file -> new KeyToPathBuilder()
					.withKey(file.getName())
					.withPath(file.getName())
					.build())
				.collect(Collectors.toList());

			hadoopConfVolume = new VolumeBuilder()
				.withName(Constants.HADOOP_CONF_VOLUME)
				.withNewConfigMap()
					.withName(getHadoopConfConfigMapName(kubernetesParameters.getClusterId()))
					.withItems(keyToPaths)
					.endConfigMap()
				.build();
		}

		final Pod podWithHadoopConf = new PodBuilder(flinkPod.getPod())
			.editOrNewSpec()
				.addNewVolumeLike(hadoopConfVolume)
					.endVolume()
				.endSpec()
			.build();

		final Container containerWithHadoopConf = new ContainerBuilder(flinkPod.getMainContainer())
			.addNewVolumeMount()
				.withName(Constants.HADOOP_CONF_VOLUME)
				.withMountPath(Constants.HADOOP_CONF_DIR_IN_POD)
				.endVolumeMount()
			.addNewEnv()
				.withName(Constants.ENV_HADOOP_CONF_DIR)
				.withValue(Constants.HADOOP_CONF_DIR_IN_POD)
				.endEnv()
			.build();

		return new FlinkPod.Builder(flinkPod)
			.withPod(podWithHadoopConf)
			.withMainContainer(containerWithHadoopConf)
			.build();
	}

	public  HadoopConfMountDecorator(AbstractKubernetesParameters kubernetesParameters);

}