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

        final PodSpec resultPodSpec =
                this.kubernetesJobManagerSpecification
                        .getDeployment()
                        .getSpec()
                        .getTemplate()
                        .getSpec();

        assertThat(resultPodSpec.getContainers()).hasSize(1);
        assertThat(resultPodSpec.getServiceAccountName()).isEqualTo(SERVICE_ACCOUNT_NAME);
        assertThat(resultPodSpec.getVolumes()).hasSize(4);

        final Container resultedMainContainer = resultPodSpec.getContainers().get(0);
        assertThat(resultedMainContainer.getName()).isEqualTo(Constants.MAIN_CONTAINER_NAME);
        assertThat(resultedMainContainer.getImage()).isEqualTo(CONTAINER_IMAGE);
        assertThat(resultedMainContainer.getImagePullPolicy())
                .isEqualTo(CONTAINER_IMAGE_PULL_POLICY.name());

        assertThat(resultedMainContainer.getEnv()).hasSize(3);
        assertThat(resultedMainContainer.getEnv().stream())
                .anyMatch(envVar -> envVar.getName().equals("key1"));

        assertThat(resultedMainContainer.getPorts()).hasSize(3);

        final Map<String, Quantity> requests = resultedMainContainer.getResources().getRequests();
        assertThat(requests.get("cpu").getAmount()).isEqualTo(Double.toString(JOB_MANAGER_CPU));
        assertThat(requests.get("memory").getAmount())
                .isEqualTo(String.valueOf(JOB_MANAGER_MEMORY));

        assertThat(resultedMainContainer.getCommand()).hasSize(1);
        // The args list is [bash, -c, 'java -classpath $FLINK_CLASSPATH ...'].
        assertThat(resultedMainContainer.getArgs()).hasSize(3);

        assertThat(resultedMainContainer.getVolumeMounts()).hasSize(4);
    }