public static KubernetesJobManagerSpecification buildKubernetesJobManagerSpecification(
            FlinkPod podTemplate, KubernetesJobManagerParameters kubernetesJobManagerParameters)
            throws IOException {
        FlinkPod flinkPod = Preconditions.checkNotNull(podTemplate).copy();
        List<HasMetadata> accompanyingResources = new ArrayList<>();

        final List<KubernetesStepDecorator> stepDecorators =
                new ArrayList<>(
                        Arrays.asList(
                                new InitJobManagerDecorator(kubernetesJobManagerParameters),
                                new EnvSecretsDecorator(kubernetesJobManagerParameters),
                                new MountSecretsDecorator(kubernetesJobManagerParameters),
                                new CmdJobManagerDecorator(kubernetesJobManagerParameters),
                                new InternalServiceDecorator(kubernetesJobManagerParameters),
                                new ExternalServiceDecorator(kubernetesJobManagerParameters)));

        Configuration configuration = kubernetesJobManagerParameters.getFlinkConfiguration();
        if (configuration.get(KUBERNETES_HADOOP_CONF_MOUNT_DECORATOR_ENABLED)) {
            stepDecorators.add(new HadoopConfMountDecorator(kubernetesJobManagerParameters));
        }
        if (configuration.get(KUBERNETES_KERBEROS_MOUNT_DECORATOR_ENABLED)) {
            stepDecorators.add(new KerberosMountDecorator(kubernetesJobManagerParameters));
        }

        stepDecorators.addAll(
                Arrays.asList(
                        new FlinkConfMountDecorator(kubernetesJobManagerParameters),
                        new PodTemplateMountDecorator(kubernetesJobManagerParameters)));

        for (KubernetesStepDecorator stepDecorator : stepDecorators) {
            flinkPod = stepDecorator.decorateFlinkPod(flinkPod);
            accompanyingResources.addAll(stepDecorator.buildAccompanyingKubernetesResources());
        }

        final Deployment deployment =
                createJobManagerDeployment(flinkPod, kubernetesJobManagerParameters);

        return new KubernetesJobManagerSpecification(deployment, accompanyingResources);
    }
@Test
    void testKerberosConfConfigMap() throws IOException {
        kubernetesJobManagerSpecification =
                KubernetesJobManagerFactory.buildKubernetesJobManagerSpecification(
                        flinkPod, kubernetesJobManagerParameters);

        final ConfigMap resultConfigMap =
                (ConfigMap)
                        this.kubernetesJobManagerSpecification.getAccompanyingResources().stream()
                                .filter(
                                        x ->
                                                x instanceof ConfigMap
                                                        && x.getMetadata()
                                                                .getName()
                                                                .equals(
                                                                        KerberosMountDecorator
                                                                                .getKerberosKrb5confConfigMapName(
                                                                                        CLUSTER_ID)))
                                .collect(Collectors.toList())
                                .get(0);

        assertThat(resultConfigMap.getApiVersion()).isEqualTo(Constants.API_VERSION);

        assertThat(resultConfigMap.getMetadata().getName())
                .isEqualTo(KerberosMountDecorator.getKerberosKrb5confConfigMapName(CLUSTER_ID));

        final Map<String, String> resultDatas = resultConfigMap.getData();
        assertThat(resultDatas).hasSize(1);
        assertThat(resultDatas.get(KRB5_CONF_FILE)).isEqualTo("some conf");
    }