private void deployJob(CuratorFramework client, final Job job) throws InterruptedException {
		if (job != null) {
			// Ensure that the path for modules used by the container to write
			// ephemeral nodes exists. The presence of this path is assumed
			// by the supervisor when it calculates stream state when it is
			// assigned leadership. See XD-2170 for details.
			try {
				client.create().creatingParentsIfNeeded().forPath(Paths.build(
						Paths.JOB_DEPLOYMENTS, job.getName(), Paths.MODULES));
			}
			catch (Exception e) {
				ZooKeeperUtils.wrapAndThrowIgnoring(e, KeeperException.NodeExistsException.class);
			}

			String statusPath = Paths.build(Paths.JOB_DEPLOYMENTS, job.getName(), Paths.STATUS);

			DeploymentUnitStatus deployingStatus = null;
			try {
				deployingStatus = new DeploymentUnitStatus(ZooKeeperUtils.bytesToMap(
						client.getData().forPath(statusPath)));
			}
			catch (Exception e) {
				// an exception indicates that the status has not been set
			}
			Assert.state(deployingStatus != null
							&& deployingStatus.getState() == DeploymentUnitStatus.State.deploying,
					String.format("Expected 'deploying' status for job '%s'; current status: %s",
							job.getName(), deployingStatus));

			ModuleDeploymentPropertiesProvider<ModuleDeploymentProperties> provider =
					new DefaultModuleDeploymentPropertiesProvider(job);

			try {
				Collection<ModuleDeploymentStatus> deploymentStatuses = new ArrayList<ModuleDeploymentStatus>();
				for (ModuleDescriptor descriptor : job.getModuleDescriptors()) {
					RuntimeModuleDeploymentProperties deploymentProperties = new RuntimeModuleDeploymentProperties();
					deploymentProperties.putAll(provider.propertiesForDescriptor(descriptor));
					Deque<Container> matchedContainers = new ArrayDeque<Container>(containerMatcher.match(descriptor,
							deploymentProperties,
							containerRepository.findAll()));
					// Modules count == 0
					if (deploymentProperties.getCount() == 0) {
						deploymentProperties.setSequence(0);
						createModuleDeploymentRequestsPath(client, descriptor, deploymentProperties);
					}
					// Modules count > 0
					else {
						for (int i = 1; i <= deploymentProperties.getCount(); i++) {
							deploymentProperties.setSequence(i);
							createModuleDeploymentRequestsPath(client, descriptor, deploymentProperties);
						}
					}
					RuntimeModuleDeploymentPropertiesProvider deploymentRuntimeProvider =
							new RuntimeModuleDeploymentPropertiesProvider(provider);

					try {
						deploymentStatuses.addAll(moduleDeploymentWriter.writeDeployment(
								descriptor, deploymentRuntimeProvider, matchedContainers));
					}
					catch (NoContainerException e) {
						logger.warn("No containers available for deployment of job {}", job.getName());
					}

					DeploymentUnitStatus status = stateCalculator.calculate(job, provider, deploymentStatuses);

					logger.info("Deployment status for job '{}': {}", job.getName(), status);

					client.setData().forPath(statusPath, ZooKeeperUtils.mapToBytes(status.toMap()));
				}
			}
			catch (InterruptedException e) {
				throw e;
			}
			catch (Exception e) {
				throw ZooKeeperUtils.wrapThrowable(e);
			}
		}
	}
Issue deployment requests for a job. This deployment will occur if:
<ul>
    <li>the job has not been destroyed</li>
    <li>the job has not been undeployed</li>
    <li>there is a container that can deploy the job</li>
</ul>

@param job the job instance to redeploy
@throws InterruptedException
spring-xd-dirt/src/main/java/org/springframework/xd/dirt/server/admin/deployment/zk/ZKJobDeploymentHandler.java