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 testAdditionalResourcesSize() throws IOException {
        kubernetesJobManagerSpecification =
                KubernetesJobManagerFactory.buildKubernetesJobManagerSpecification(
                        flinkPod, kubernetesJobManagerParameters);

        final List<HasMetadata> resultAdditionalResources =
                this.kubernetesJobManagerSpecification.getAccompanyingResources();
        assertThat(resultAdditionalResources).hasSize(5);

        final List<HasMetadata> resultServices =
                resultAdditionalResources.stream()
                        .filter(x -> x instanceof Service)
                        .collect(Collectors.toList());
        assertThat(resultServices).hasSize(2);

        final List<HasMetadata> resultConfigMaps =
                resultAdditionalResources.stream()
                        .filter(x -> x instanceof ConfigMap)
                        .collect(Collectors.toList());
        assertThat(resultConfigMaps).hasSize(2);

        final List<HasMetadata> resultSecrets =
                resultAdditionalResources.stream()
                        .filter(x -> x instanceof Secret)
                        .collect(Collectors.toList());
        assertThat(resultSecrets).hasSize(1);
    }