@Override
    public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException {
        if (!kubernetesJobManagerParameters.isInternalServiceEnabled()) {
            return Collections.emptyList();
        }

        final String serviceName =
                getInternalServiceName(kubernetesJobManagerParameters.getClusterId());

        final Service headlessService =
                HeadlessClusterIPService.INSTANCE.buildUpInternalService(
                        kubernetesJobManagerParameters);

        // Set job manager address to namespaced service name
        final String namespace = kubernetesJobManagerParameters.getNamespace();
        kubernetesJobManagerParameters
                .getFlinkConfiguration()
                .set(
                        JobManagerOptions.ADDRESS,
                        getNamespacedInternalServiceName(serviceName, namespace));

        return Collections.singletonList(headlessService);
    }
@Test
    void testBuildAccompanyingKubernetesResources() throws IOException {
        final List<HasMetadata> resources =
                this.internalServiceDecorator.buildAccompanyingKubernetesResources();
        assertThat(resources).hasSize(1);

        assertThat(InternalServiceDecorator.getNamespacedInternalServiceName(CLUSTER_ID, NAMESPACE))
                .isEqualTo(this.flinkConfig.get(JobManagerOptions.ADDRESS));

        final Service internalService = (Service) resources.get(0);

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

        assertThat(internalService.getMetadata().getName())
                .isEqualTo(InternalServiceDecorator.getInternalServiceName(CLUSTER_ID));

        final Map<String, String> expectedLabels = getCommonLabels();
        assertThat(internalService.getMetadata().getLabels()).isEqualTo(expectedLabels);

        assertThat(internalService.getMetadata().getAnnotations()).isEqualTo(userAnnotations);

        assertThat(internalService.getSpec().getType()).isNull();
        assertThat(internalService.getSpec().getClusterIP()).isEqualTo("None");

        List<ServicePort> expectedServicePorts =
                Arrays.asList(
                        new ServicePortBuilder()
                                .withName(Constants.JOB_MANAGER_RPC_PORT_NAME)
                                .withPort(RPC_PORT)
                                .build(),
                        new ServicePortBuilder()
                                .withName(Constants.BLOB_SERVER_PORT_NAME)
                                .withPort(BLOB_SERVER_PORT)
                                .build());
        assertThat(internalService.getSpec().getPorts()).isEqualTo(expectedServicePorts);

        expectedLabels.put(Constants.LABEL_COMPONENT_KEY, Constants.LABEL_COMPONENT_JOB_MANAGER);
        assertThat(internalService.getSpec().getSelector()).isEqualTo(expectedLabels);
    }