@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 testDisableInternalService() throws IOException {
        this.flinkConfig.set(
                HighAvailabilityOptions.HA_MODE, HighAvailabilityMode.ZOOKEEPER.name());

        final List<HasMetadata> resources =
                this.internalServiceDecorator.buildAccompanyingKubernetesResources();
        assertThat(resources).isEmpty();
    }