@Override
		@SuppressWarnings("rawtypes")
		public void takeLeadership(CuratorFramework client) throws Exception {
			logger.info("Leader Admin {} is watching for stream/job deployment requests.", getId());
			PathChildrenCache containers = null;
			PathChildrenCache streamDeployments = null;
			PathChildrenCache jobDeployments = null;
			PathChildrenCache moduleDeploymentRequests = null;
			ContainerListener containerListener;

			try {
				String requestedModulesPath = Paths.build(Paths.MODULE_DEPLOYMENTS, Paths.REQUESTED);
				Paths.ensurePath(client, requestedModulesPath);
				String allocatedModulesPath = Paths.build(Paths.MODULE_DEPLOYMENTS, Paths.ALLOCATED);
				Paths.ensurePath(client, allocatedModulesPath);
				moduleDeploymentRequests = instantiatePathChildrenCache(client, requestedModulesPath);
				moduleDeploymentRequests.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);

				streamDeployments = instantiatePathChildrenCache(client, Paths.STREAM_DEPLOYMENTS);
//				// using BUILD_INITIAL_CACHE so that all known streams are populated
//				// in the cache before invoking recalculateStreamStates; same for
//				// jobs below
				streamDeployments.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);

				jobDeployments = instantiatePathChildrenCache(client, Paths.JOB_DEPLOYMENTS);
				jobDeployments.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);

				SupervisorElectedEvent supervisorElectedEvent = new SupervisorElectedEvent(moduleDeploymentRequests,
						streamDeployments, jobDeployments);

				Map<String, SupervisorElectionListener> listenersMap =
						applicationContext.getBeansOfType(SupervisorElectionListener.class);
				for (Map.Entry<String, SupervisorElectionListener> entry : listenersMap.entrySet()) {
					entry.getValue().onSupervisorElected(supervisorElectedEvent);
				}

				containerListener = new ContainerListener(zkConnection,
						containerRepository,
						streamFactory,
						jobFactory,
						streamDeployments,
						jobDeployments,
						moduleDeploymentRequests,
						containerMatcher,
						moduleDeploymentWriter,
						stateCalculator,
						executorService,
						quietPeriod);

				containers = instantiatePathChildrenCache(client, Paths.CONTAINERS);
				containers.getListenable().addListener(containerListener);
				containers.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

				containerListener.scheduleDepartedContainerDeployer();

				deploymentQueueForConsumer = new DeploymentQueue(client, deploymentMessageConsumer,
						Paths.DEPLOYMENT_QUEUE, executorService);
				deploymentQueueForConsumer.start();

				Thread.sleep(Long.MAX_VALUE);
			}
			catch (InterruptedException e) {
				logger.info("Leadership canceled due to thread interrupt");
				Thread.currentThread().interrupt();
			}
			finally {
				if (containers != null) {
					containers.close();
				}

				if (streamDeployments != null) {
					streamDeployments.close();
				}

				if (jobDeployments != null) {
					jobDeployments.close();
				}
				if (moduleDeploymentRequests != null) {
					moduleDeploymentRequests.close();
				}
				if (deploymentQueueForConsumer != null) {
					try {
						deploymentQueueForConsumer.destroy();
					}
					catch (IOException e) {
						logger.warn("Exception closing the distributed queue producer " + e);
					}
				}
			}
		}
{@inheritDoc}
<p/>
Upon leadership election, this Admin server will create a {@link PathChildrenCache}
for {@link Paths#STREAMS} and {@link Paths#JOBS}. These caches will have
{@link PathChildrenCacheListener PathChildrenCacheListeners} attached to them
that will react to stream and job creation and deletion. Upon leadership
relinquishment, the listeners will be removed and the caches shut down.
spring-xd-dirt/src/main/java/org/springframework/xd/dirt/server/admin/deployment/zk/DeploymentSupervisor.java