class KubernetesJobManagerFactory {

	public static KubernetesJobManagerSpecification buildKubernetesJobManagerSpecification(
			KubernetesJobManagerParameters kubernetesJobManagerParameters) throws IOException {
		FlinkPod flinkPod = new FlinkPod.Builder().build();
		List<HasMetadata> accompanyingResources = new ArrayList<>();

		final KubernetesStepDecorator[] stepDecorators = new KubernetesStepDecorator[] {
			new InitJobManagerDecorator(kubernetesJobManagerParameters),
			new EnvSecretsDecorator(kubernetesJobManagerParameters),
			new MountSecretsDecorator(kubernetesJobManagerParameters),
			new JavaCmdJobManagerDecorator(kubernetesJobManagerParameters),
			new InternalServiceDecorator(kubernetesJobManagerParameters),
			new ExternalServiceDecorator(kubernetesJobManagerParameters),
			new HadoopConfMountDecorator(kubernetesJobManagerParameters),
			new FlinkConfMountDecorator(kubernetesJobManagerParameters)};

		for (KubernetesStepDecorator stepDecorator: stepDecorators) {
			flinkPod = stepDecorator.decorateFlinkPod(flinkPod);
			accompanyingResources.addAll(stepDecorator.buildAccompanyingKubernetesResources());
		}

		final Deployment deployment = createJobManagerDeployment(flinkPod, kubernetesJobManagerParameters);

		return new KubernetesJobManagerSpecification(deployment, accompanyingResources);
	}

}