class InternalServiceDecorator extends AbstractKubernetesStepDecorator {

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

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

		final Service headlessService = new ServiceBuilder()
			.withApiVersion(Constants.API_VERSION)
			.withNewMetadata()
				.withName(serviceName)
				.withLabels(kubernetesJobManagerParameters.getCommonLabels())
				.endMetadata()
			.withNewSpec()
				.withClusterIP(Constants.HEADLESS_SERVICE_CLUSTER_IP)
				.withSelector(kubernetesJobManagerParameters.getLabels())
				.addNewPort()
					.withName(Constants.JOB_MANAGER_RPC_PORT_NAME)
					.withPort(kubernetesJobManagerParameters.getRPCPort())
					.endPort()
				.addNewPort()
					.withName(Constants.BLOB_SERVER_PORT_NAME)
					.withPort(kubernetesJobManagerParameters.getBlobServerPort())
					.endPort()
				.endSpec()
			.build();

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

		return Collections.singletonList(headlessService);
	}

	public  InternalServiceDecorator(KubernetesJobManagerParameters kubernetesJobManagerParameters);

}