@Override
    public void createJobManagerComponent(KubernetesJobManagerSpecification kubernetesJMSpec) {
        final Deployment deployment = kubernetesJMSpec.getDeployment();
        final List<HasMetadata> accompanyingResources = kubernetesJMSpec.getAccompanyingResources();

        // create Deployment
        LOG.debug(
                "Start to create deployment with spec {}{}",
                System.lineSeparator(),
                KubernetesUtils.tryToGetPrettyPrintYaml(deployment));
        final Deployment createdDeployment = this.internalClient.resource(deployment).create();

        // Note that we should use the uid of the created Deployment for the OwnerReference.
        setOwnerReference(createdDeployment, accompanyingResources);

        this.internalClient.resourceList(accompanyingResources).createOrReplace();
    }
@Test
    void testCreateFlinkMasterComponent() throws Exception {
        flinkKubeClient.createJobManagerComponent(this.kubernetesJobManagerSpecification);

        final List<Deployment> resultedDeployments =
                kubeClient.apps().deployments().inNamespace(NAMESPACE).list().getItems();
        assertThat(resultedDeployments).hasSize(1);

        final List<ConfigMap> resultedConfigMaps =
                kubeClient.configMaps().inNamespace(NAMESPACE).list().getItems();
        assertThat(resultedConfigMaps).hasSize(1);

        final List<Service> resultedServices =
                kubeClient.services().inNamespace(NAMESPACE).list().getItems();
        assertThat(resultedServices).hasSize(2);

        testOwnerReferenceSetting(resultedDeployments.get(0), resultedConfigMaps);
        testOwnerReferenceSetting(resultedDeployments.get(0), resultedServices);
    }