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

        final List<Service> resultServices =
                this.kubernetesJobManagerSpecification.getAccompanyingResources().stream()
                        .filter(x -> x instanceof Service)
                        .map(x -> (Service) x)
                        .collect(Collectors.toList());

        assertThat(resultServices).hasSize(2);

        final List<Service> internalServiceCandidates =
                resultServices.stream()
                        .filter(
                                x ->
                                        x.getMetadata()
                                                .getName()
                                                .equals(
                                                        InternalServiceDecorator
                                                                .getInternalServiceName(
                                                                        CLUSTER_ID)))
                        .collect(Collectors.toList());
        assertThat(internalServiceCandidates).hasSize(1);

        final List<Service> restServiceCandidates =
                resultServices.stream()
                        .filter(
                                x ->
                                        x.getMetadata()
                                                .getName()
                                                .equals(
                                                        ExternalServiceDecorator
                                                                .getExternalServiceName(
                                                                        CLUSTER_ID)))
                        .collect(Collectors.toList());
        assertThat(restServiceCandidates).hasSize(1);

        final Service resultInternalService = internalServiceCandidates.get(0);
        assertThat(resultInternalService.getMetadata().getLabels()).hasSize(2);

        assertThat(resultInternalService.getSpec().getType()).isNull();
        assertThat(resultInternalService.getSpec().getClusterIP())
                .isEqualTo(HeadlessClusterIPService.HEADLESS_CLUSTER_IP);
        assertThat(resultInternalService.getSpec().getPorts()).hasSize(2);
        assertThat(resultInternalService.getSpec().getSelector()).hasSize(3);

        final Service resultRestService = restServiceCandidates.get(0);
        assertThat(resultRestService.getMetadata().getLabels()).hasSize(2);

        assertThat(resultRestService.getSpec().getType()).isEqualTo("ClusterIP");
        assertThat(resultRestService.getSpec().getPorts()).hasSize(1);
        assertThat(resultRestService.getSpec().getSelector()).hasSize(3);
    }