@Override
	protected void deployModules(Container container) throws Exception {
		CuratorFramework client = getClient();
		if (client.getState() == CuratorFrameworkState.STOPPED) {
			return;
		}

		// the departed container may have hosted multiple modules
		// for the same stream; therefore each stream that is loaded
		// will be cached to avoid reloading for each module
		Map<String, Stream> streamMap = new HashMap<String, Stream>();

		String containerDeployments = Paths.build(Paths.MODULE_DEPLOYMENTS, Paths.ALLOCATED, container.getName());
		List<String> deployments = client.getChildren().forPath(containerDeployments);

		// Stream modules need to be deployed in the correct order;
		// this is done in a two-pass operation: gather, then re-deploy.
		Set<ModuleDeployment> streamModuleDeployments = new TreeSet<ModuleDeployment>();
		for (String deployment : deployments) {
			ModuleDeploymentsPath moduleDeploymentsPath =
					new ModuleDeploymentsPath(Paths.build(containerDeployments, deployment));

			// reuse the module deployment properties used to deploy
			// this module; this may contain properties specific to
			// the container that just departed (such as partition
			// index in the case of partitioned streams)
			RuntimeModuleDeploymentProperties deploymentProperties = new RuntimeModuleDeploymentProperties();
			deploymentProperties.putAll(ZooKeeperUtils.bytesToMap(client.getData().forPath(
					moduleDeploymentsPath.build())));

			String unitName = moduleDeploymentsPath.getDeploymentUnitName();
			String moduleType = moduleDeploymentsPath.getModuleType();

			if (ModuleType.job.toString().equals(moduleType)) {
				try {
					final Job job = DeploymentLoader.loadJob(client, unitName, jobFactory);
					if (job != null) {
						redeployModule(new ModuleDeployment(job, job.getJobModuleDescriptor(),
								deploymentProperties), false);
					}
				}
				catch (Exception e) {
					logger.error(String.format("Exception loading job %s", unitName), e);
				}
			}
			else {
				Stream stream = streamMap.get(unitName);
				if (stream == null) {
					try {
						stream = DeploymentLoader.loadStream(client, unitName, streamFactory);
						streamMap.put(unitName, stream);
					}
					catch (Exception e) {
						logger.error(String.format("Exception loading stream %s.", unitName), e);
					}
				}
				if (stream != null) {
					streamModuleDeployments.add(new ModuleDeployment(stream,
							stream.getModuleDescriptor(moduleDeploymentsPath.getModuleLabel()),
							deploymentProperties));
				}
			}
		}

		for (ModuleDeployment moduleDeployment : streamModuleDeployments) {
			redeployModule(moduleDeployment, false);
		}

		// remove the deployments from the departed container
		client.delete().deletingChildrenIfNeeded().forPath(
				Paths.build(Paths.MODULE_DEPLOYMENTS, Paths.ALLOCATED, container.getName()));
	}
Handle the departure of a container. This will scan the list of modules
deployed to the departing container and redeploy them if required.

@param container the container that departed
spring-xd-dirt/src/main/java/org/springframework/xd/dirt/server/admin/deployment/zk/DepartedContainerModuleRedeployer.java