@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.getPodWithoutMainContainer())
                        .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();
    }
@Test
    void testMainContainerWithHadoopConfVolumeMount() throws IOException {
        setHadoopConfDirEnv();
        generateHadoopConfFileItems();
        final FlinkPod resultFlinkPod = hadoopConfMountDecorator.decorateFlinkPod(baseFlinkPod);

        final List<VolumeMount> resultVolumeMounts =
                resultFlinkPod.getMainContainer().getVolumeMounts();
        assertThat(resultVolumeMounts).hasSize(1);
        final VolumeMount resultVolumeMount = resultVolumeMounts.get(0);
        assertThat(resultVolumeMount.getName()).isEqualTo(Constants.HADOOP_CONF_VOLUME);
        assertThat(resultVolumeMount.getMountPath()).isEqualTo(Constants.HADOOP_CONF_DIR_IN_POD);

        final Map<String, String> expectedEnvs =
                new HashMap<String, String>() {
                    {
                        put(Constants.ENV_HADOOP_CONF_DIR, Constants.HADOOP_CONF_DIR_IN_POD);
                    }
                };
        final Map<String, String> resultEnvs =
                resultFlinkPod.getMainContainer().getEnv().stream()
                        .collect(Collectors.toMap(EnvVar::getName, EnvVar::getValue));
        assertThat(resultEnvs).isEqualTo(expectedEnvs);
    }