First time, synchronizes and makes sure all nodes in the path are created. Subsequent calls
with this instance are NOPs.

@param client ZK client
@throws Exception ZK errors
559003
public void ensure(CuratorZookeeperClient client) throws Exception
    {
        Helper localHelper = helper.get();
        localHelper.ensure(client, path, makeLastNode);
    }
460471
public void init() throws IOException {
    try {
      if (zooKeeper.exists(zooKeeperParentPath, false) == null) {
        zooKeeper.create(zooKeeperParentPath, new byte[] { '0' },
            ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        LOG.info("Created ZNode " + zooKeeperParentPath);
      }
      if (zooKeeper.exists(ledgerParentPath, false) == null) {
        zooKeeper.create(ledgerParentPath, new byte[] { '0' },
            ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        LOG.info("Created ZNode" + ledgerParentPath);
      }
    } catch (InterruptedException e) {
      interruptedException("Interrupted ensuring that ZNodes " +
          zooKeeperParentPath + " and " + ledgerParentPath + " exist!", e);
    } catch (KeeperException e) {
      keeperException(
          "Unrecoverable ZooKeeper error ensuring that ZNodes "
              + zooKeeperParentPath + " and " + ledgerParentPath + " exist!",
          e);
    }
  }
771570
public void commit() throws Exception {
        final Map<String, Object> params = new HashMap<String, Object>();

        final String applicationId = getPropertyAsString("applicationId");
        if (applicationId != null) params.put("applicationId", applicationId);

        final String name = getName();
        if (name != null) params.put("name", name);

        final String fallbackNumber = getFallbackNumber();
        if (fallbackNumber != null) params.put("fallbackNumber", fallbackNumber);

        final String uri = getUri();
        client.post(uri, params);

        final JSONObject object = toJSONObject(client.get(uri, null));
        updateProperties(object);
    }
136238
void start() throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "start");

        synchronized (stateLock) {

            // We will start each consumer inside its own catch block so that
            // the failing of a single consumer does not adversely affect any
            // other. The first exception that is received is stored in this
            // variable and will be passed back up to the Connection when all
            // the starts are complete.
            JMSException excReceived = null;

            // Start all the synchronous consumers

            // Take a copy of the list of synchronous consumers. A sync block in
            // createConsumer means that you cannot add anything to this list until
            // session.start completes, however consumers can still be removed (as
            // a result of the consumer.close operation), so by the time we get to
            // calling start on the elements in this list, some of them may already
            // be closed (pathalogical scenario). Calling start on a closed consumer
            // has no effect, so this shouldn't cause a problem.
            Object[] syncList = syncConsumers.toArray();

            for (int i = 0; i < syncList.length; i++) {
                try {
                    ((JmsMsgConsumerImpl) syncList[i]).start();
                } catch (JMSException e) {
                    // No FFDC code needed
                    // Cache the exception to be thrown back at the end of the method,
                    // but don't take any other action here. The exception will already
                    // have been logged in JmsMsgConsumerImpl.start.
                    if (excReceived == null) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                            SibTr.debug(this, tc, "JMSException received at Session.start. Will continue trying to start the remaining consumers.");
                        // Cache this exception for later use.
                        excReceived = e;
                    }
                }
            }

            // Start all the asynchronous consumers.
            Iterator asyncList = asyncConsumers.iterator();
            while (asyncList.hasNext()) {
                try {
                    ((JmsMsgConsumerImpl) asyncList.next()).start();
                } catch (JMSException e) {
                    // No FFDC code needed

                    // Cache the exception to be thrown back at the end of the method,
                    // but don't take any other action here. The exception will already
                    // have been logged in JmsMsgConsumerImpl.start.
                    if (excReceived == null) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                            SibTr.debug(this, tc, "JMSException received at Session.start. Will continue trying to start the remaining asynch consumers.");
                        // Cache this exception for later use.
                        excReceived = e;
                    }
                }
            }

            // Check whether an exception was thrown at any point during this method.
            if (excReceived != null) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(this, tc, "An exception was received during Session.start. Propogate to Connection.");
                throw excReceived;
            }

            // Flag that we have started everything.
            setState(STARTED);

        } // Release the lock

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "start");
    }
905283
public static void createEphemeralPath(ZkClient zkClient, String path, String data) {
        try {
            zkClient.createEphemeral(path, Utils.getBytes(data));
        } catch (ZkNoNodeException e) {
            createParentPath(zkClient, path);
            zkClient.createEphemeral(path, Utils.getBytes(data));
        }
    }
253191
public synchronized void create( HttpServletRequest request, HttpServletResponse response, ServletContext servletContext )
        throws Exception
    {
        reinitialize( request, response, servletContext );
        JavaControlUtils.initJavaControls( request, response, servletContext, this );
        onCreate();
    }
276235
public void releaseAll() throws Exception {
		Collection<String> children = getAllPaths();

		Exception exception = null;

		for (String child: children) {
			try {
				release(child);
			} catch (Exception e) {
				exception = ExceptionUtils.firstOrSuppressed(e, exception);
			}
		}

		if (exception != null) {
			throw new Exception("Could not properly release all state nodes.", exception);
		}
	}
517439
@Override
    public void clientCreated(Client client) {
        if (AGENT_PORT_TYPE.equals(getEndpointName(client.getEndpoint()))) {
            return;
        }
        processStart(client.getEndpoint(), EventTypeEnum.CLIENT_CREATE);
    }
924457
private void connect(ZooKeeperClient injectedClient) throws InterruptedException {
        if (!initialized.get()) {
            if (injectedClient == null) {
                List<InetSocketAddress> hosts = new ArrayList<InetSocketAddress>();
                for (String host : config.hosts.split(",")) {
                    String[] parts = host.split(":");
                    try {
                        hosts.add(new InetSocketAddress(parts[0], Integer.parseInt(parts[1])));
                    } catch (Exception e) {
                        LOG.error("Invalid ZK host '"+host+"', need to skip, problem: "+e.getMessage());
                    }
                }
                // TODO: What about empty hosts at this point?
                LOG.info("Connecting to hosts: {}", hosts.toString());
                injectedClient = new ZooKeeperClient(Amount.of((int) config.zkTimeout, Time.MILLISECONDS),
                        hosts);
            }
            zk = injectedClient;
            claimer.start();
            LOG.info("Registering connection watcher.");
            zk.register(connectionWatcher);
        }
        // and then see that we can actually get the client without problems
        try {
            zk.get();
        } catch (ZooKeeperConnectionException e) {
            throw ZKException.from(e);
        }
    }
224034
@FFDCIgnore(value = { InterruptedException.class })
    public void checkinClient(Client client) {
        boolean done = false;

        //If no more space. Return. Shouldn't be checking more than we have. Shouldn't happen besides unit test anyways.
        if (clients.remainingCapacity() == 0) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, "Checking in more clients than there should be.");
            }
            return;
        }

        while (!done) {
            try {
                //check if this client was already checked-in.
                if (!clients.contains(client)) {
                    clients.putFirst(client);
                    if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                        int clientHash = (client != null) ? client.hashCode() : 0;
                        Tr.event(tc, "post-checkin  - (" + clientHash + ") " + (numClients - clients.remainingCapacity()) + " of " + numClients + " clients available.");
                    }
                }
                done = true;
            } catch (InterruptedException e) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
                    Tr.event(tc, "InterruptedException during checkin - will retry");
            }
        }

    }
39788
public void init(){
    try {
      zk = new ZooKeeper(parent.getProperties().get(TeknekDaemon.ZK_SERVER_LIST).toString(), 30000, this);
    } catch (IOException e1) {
      throw new RuntimeException(e1);
    }
    Feed feed = DriverFactory.buildFeed(plan.getFeedDesc());
    List<WorkerStatus> workerStatus;
    try {
      workerStatus = parent.getWorkerDao().findAllWorkerStatusForPlan(plan, otherWorkers);
    } catch (WorkerDaoException e1) {
      throw new RuntimeException(e1);
    }
    if (workerStatus.size() >= feed.getFeedPartitions().size()){
      throw new RuntimeException("Number of running workers " + workerStatus.size()+" >= feed partitions " + feed.getFeedPartitions().size() +" plan should be fixed " +plan.getName());
    }
    FeedPartition toProcess = findPartitionToProcess(workerStatus, feed.getFeedPartitions());
    if (toProcess != null){
      driver = DriverFactory.createDriver(toProcess, plan, parent.getMetricRegistry());
      driver.initialize(); 
      WorkerStatus iGotThis = new WorkerStatus(myId.toString(), toProcess.getPartitionId() , parent.getMyId());
      try {
        parent.getWorkerDao().registerWorkerStatus(zk, plan, iGotThis);
      } catch (WorkerDaoException e) {
        throw new RuntimeException(e);
      }
    } else {
      throw new RuntimeException("Could not start plan "+plan.getName());
    }
  }
1286810
protected void doPost(String path) throws ClientException {
        this.readLock.lock();
        try {
            ClientResponse response = getResourceWrapper().rewritten(path, HttpMethod.POST)
                    .post(ClientResponse.class);
            errorIfStatusNotEqualTo(response, ClientResponse.Status.OK, ClientResponse.Status.NO_CONTENT);
            response.close();
        } catch (ClientHandlerException ex) {
            throw new ClientException(ClientResponse.Status.INTERNAL_SERVER_ERROR, ex.getMessage());
        } finally {
            this.readLock.unlock();
        }
    }
1040271
@Override
    protected void doStart() throws Exception
    {
        log.info("starting...");
        super.doStart();
        _client = newClient(_serverString);
        if (_client == null)
        {
            throw new IllegalStateException("newClient(" + _serverString + ") returns null.");
        }
        log.info("use " + _client.getClass().getSimpleName() + " as client factory.");
        _client.establish();
        log.info("started.");
    }
271962
public void start()
    throws Exception
  {
    if (_port < 0 && _unixPath == null) {
      return;
    }

    if (! _lifecycle.toStarting())
      return;

    boolean isValid = false;
    try {
      bind();
      postBind();

      enable();

      _acceptTask = new AcceptTcp(this, _serverSocket);
      _threadPool.execute(_acceptTask);

      // _connThreadPool.start();

      _suspendAlarm = new Alarm(new SuspendReaper());
      _suspendAlarm.runAfter(_suspendReaperTimeout);

      isValid = true;
    } finally {
      if (! isValid) {
        close();
      }
    }
  }
1341544
@Override
    public final void exception ( Exception e ) {
        this.error = true;
        this.exception = e;
        this.received = true;
        synchronized ( this ) {
            notifyAll();
        }
    }
1068174
@Override
    protected void postInit() throws Exception {
        if (controlTree == null) {
            throw new RuntimeException("controlTree not set");
        }

        controlTree.create(NODES, false);
        controlTree.create(LEADERS, false);

        if (controlTree.exists(myNodeInfo.treeNodePath)) {
            LOG.error("A node with the name " + myNodeInfo.getName() + " already exists!");
            throw new RuntimeException("Initialization failure");
        }

        LOG.info("Required peer node properties: {}", requiredPeerNodeProperties);
        LOG.info("Required server properties: {}", requiredServerProperties);

        final Set<String> requiredProperties = (myId == 0 ? requiredServerProperties : requiredPeerNodeProperties);
        for (String property : requiredProperties) {
            if (!property.equals("id") && myNodeInfo.get(property) == null) {
                LOG.error("Required property {} not set!", property);
                throw new RuntimeException("Initialization failure");
            }
        }

        /// the calls in the demarcated section need to be in this specific order to avoid a possible race between nodes and leaders
        // `>>>> BEGIN CAREFULLY ORDERED SECTION
        branch = new DistributedBranchHelper(controlTree, NODES, false) {
            @Override
            protected boolean isNodeComplete(String node, Set<String> properties) {
                if (!properties.contains("id")) {
                    return false;
                }
                final short id = Short.parseShort(new String(controlTree.get(node + "/id"), Charsets.UTF_8));
                final Set<String> requiredProperties = (id == 0 ? requiredServerProperties : requiredPeerNodeProperties);
                final boolean success = properties.containsAll(requiredProperties);
                return success;
            }
        };

        branch.addListener(new DistributedTree.ListenerAdapter() {
            @Override
            public void nodeChildAdded(String parentPath, String childName) {
                AbstractCluster.this.nodeAdded(childName);
            }

            @Override
            public void nodeChildDeleted(String parentPath, String childName) {
                AbstractCluster.this.nodeRemoved(childName);
            }
        });
        // This read and handles the nodes.
        branch.init();

        controlTree.addListener(LEADERS, new DistributedTree.ListenerAdapter() {
            @Override
            public void nodeChildAdded(String parentPath, String childName) {
                AbstractCluster.this.leaderAdded(childName);
            }

            @Override
            public void nodeChildDeleted(String parentPath, String childName) {
                AbstractCluster.this.leaderRemoved(childName);
            }
        });

        myNodeInfo.writeToTree();

        setReady(true);

        super.postInit();

        joined = true;
        fireJoinedCluster();
        for (short id : masters.keySet()) {
            if (id != myId)
                fireNodeAdded(id);
        }
    }
1231746
@Override
    public void start() throws Exception {
        executor = Executors.newCachedThreadPool(
                new ThreadFactoryBuilder().setNameFormat("KafkaConsumer-%d").build());
        connector = Consumer.createJavaConsumerConnector(new ConsumerConfig(consumerProps));

        final Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(ImmutableMap.of(topic, readers));

        final List<KafkaStream<byte[], byte[]>> streamList = streams.get(topic);
        if (streamList == null) {
            throw new RuntimeException(topic + " is not valid");
        }

        running = true;
        for (KafkaStream<byte[], byte[]> stream : streamList) {
            final ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
            runners.add(
                    executor.submit(new Runnable() {
                        @Override
                        public void run() {
                            while (running) {
                                try {
                                    long pause = Math.min(pausedTime.get(), MAX_PAUSE);
                                    if (pause > 0) {
                                        Thread.sleep(pause);
                                        pausedTime.set(0);
                                    }
                                    byte[] message = iterator.next().message();
                                    router.process(
                                            KafkaConsumer.this,
                                            new DefaultMessageContainer(new Message(topic, message), jsonMapper));
                                } catch (ConsumerTimeoutException timeoutException) {
                                    // do nothing
                                } catch (Exception e) {
                                    log.error("Exception on consuming kafka with topic: " + topic, e);
                                }
                            }
                        }
                    })
            );
        }
    }