Ensure required zk node is present in zookeeper.
2938
@Synchronized
    @SneakyThrows(Exception.class)
    private void tryInit() {
        if (!zkInit) {
            // we are making remote calls under a lock but this is only done for initialization at 
            // the start of controller process.
            ZKUtils.createPathIfNotExists(zkClient, zkPath, HostContainerMap.EMPTY.toBytes());
            hostContainerMapNode.getListenable().addListener(this::updateMap);
            hostContainerMapNode.start(true);
            updateMap();
            zkInit = true;
        }
    }
333620
def find_existing_master
      if data = @zk.get(redis_nodes_path).first
        nodes = symbolize_keys(decode(data))
        master = node_from(nodes[:master])
        logger.info("Master from existing znode config: #{master || 'none'}")
        # Check for case where a node previously thought to be the master was
        # somehow manually reconfigured to be a slave outside of the node manager's
        # control.
        begin
          if master && master.slave?
            raise InvalidNodeRoleError.new(master, :master, :slave)
          end
        rescue RedisFailover::NodeUnavailableError => ex
          logger.warn("Failed to check whether existing master has invalid role: #{ex.inspect}")
        end

        master
      end
    rescue ZK::Exceptions::NoNode
      # blank slate, no last known master
      nil
    end
319829
private static void registerHttpAddress(AvatarZooKeeperClient zk, Configuration originalConf,
      Configuration conf, boolean toOverwrite) throws UnsupportedEncodingException, IOException {
    LOG.info("Updating Http Address information in ZooKeeper");
    String addr = conf.get(FSConstants.DFS_NAMENODE_HTTP_ADDRESS_KEY);
    String defaultAddr = originalConf
        .get(FSConstants.DFS_NAMENODE_HTTP_ADDRESS_KEY);
    registerSocketAddress(zk, defaultAddr, addr, toOverwrite);
    registerAliases(zk, conf, FSConstants.DFS_HTTP_ALIASES, addr, toOverwrite);
  }
894389
def setup(self):
        '''
        
        '''
        self.zoo_client.add_listener(self.state_listener)

        if self.ensure:
            self.zoo_client.ensure_path(self.my_file)
1149356
int claimResource(ZooKeeper zookeeper, String poolNode, int poolSize)
            throws KeeperException, InterruptedException {

        logger.debug("Trying to claim a resource.");
        List<String> claimedResources = zookeeper.getChildren(poolNode, false);
        if (claimedResources.size() >= poolSize) {
            logger.debug("No resources available at the moment (pool size: {}), waiting.", poolSize);
            // No resources available. Wait for a resource to become available.
            final CountDownLatch latch = new CountDownLatch(1);
            zookeeper.getChildren(poolNode, event -> latch.countDown());
            latch.await();
            return claimResource(zookeeper, poolNode, poolSize);
        }

        // Try to claim an available resource.
        for (int i = 0; i < poolSize; i++) {
            String resourcePath = Integer.toString(i);
            if (!claimedResources.contains(resourcePath)) {
                String node;
                try {
                    logger.debug("Trying to claim seemingly available resource {}.", resourcePath);
                    node = zookeeper.create(
                            poolNode + "/" + resourcePath,
                            new byte[0],
                            ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.EPHEMERAL
                    );
                } catch (KeeperException e) {
                    if (e.code() == KeeperException.Code.NODEEXISTS) {
                        // Failed to claim this resource for some reason.
                        continue;
                    } else {
                        // Unexpected failure.
                        throw e;
                    }
                }

                zookeeper.exists(node, event -> {
                    if (event.getType() == EventType.NodeDeleted) {
                        // Invalidate our claim when the node is deleted by some other process.
                        logger.debug("Resource-claim node unexpectedly deleted ({})", resource);
                        close(true);
                    }
                });

                logger.debug("Successfully claimed resource {}.", resourcePath);
                return i;
            }
        }

        return claimResource(zookeeper, poolNode, poolSize);
    }
802920
public void createZookeeperBase() throws WorkerDaoException {
    createPersistentEmptyNodeIfNotExist(BASE_ZK);
    createPersistentEmptyNodeIfNotExist(WORKERS_ZK);
    createPersistentEmptyNodeIfNotExist(PLANS_ZK);
    createPersistentEmptyNodeIfNotExist(SAVED_ZK);
    createPersistentEmptyNodeIfNotExist(LOCKS_ZK);
    createPersistentEmptyNodeIfNotExist(PLAN_WORKERS_ZK);
  }
1062803
public void resumeZooKeeper() throws Exception {
        val zk = new ZooKeeperServiceRunner(this.zkPort, this.secureZK, this.tLSKeyStore, this.tLSKeyStorePasswordPath, this.tlsTrustStore);
        if (this.zkServer.compareAndSet(null, zk)) {
            // Initialize ZK runner (since nobody else did it for us).
            zk.initialize();
            log.info("ZooKeeper initialized.");
        } else {
            zk.close();
        }

        // Start or resume ZK.
        this.zkServer.get().start();
        log.info("ZooKeeper resumed.");
    }
113079
private static boolean registerClientProtocolAddress(AvatarZooKeeperClient zk,
      Configuration originalConf, Configuration conf, boolean toOverwrite)
      throws UnsupportedEncodingException, IOException {
    LOG.info("Updating Client Address information in ZooKeeper");
    InetSocketAddress addr = NameNode.getClientProtocolAddress(conf);
    if (addr == null) {
      LOG.error(FSConstants.DFS_NAMENODE_RPC_ADDRESS_KEY
          + " for primary service is not defined");
      return true;
    }
    InetSocketAddress defaultAddr = NameNode.getClientProtocolAddress(originalConf);
    if (defaultAddr == null) {
      LOG.error(FSConstants.DFS_NAMENODE_RPC_ADDRESS_KEY
          + " for default service is not defined");
      return true;
    }
    registerSocketAddress(zk,
        originalConf.get(NameNode.DFS_NAMENODE_RPC_ADDRESS_KEY),
        conf.get(NameNode.DFS_NAMENODE_RPC_ADDRESS_KEY), toOverwrite);
    /** TODO later: need to handle alias leave it as it is now */
    registerAliases(zk, conf, FSConstants.FS_NAMENODE_ALIASES,
        conf.get(NameNode.DFS_NAMENODE_RPC_ADDRESS_KEY), toOverwrite);
    return false;
  }
876586
void enterRestore() {
        createZKDirectory(VoltZK.restore);
        createZKDirectory(VoltZK.restore_barrier);
        createZKDirectory(VoltZK.restore_barrier2);

        try {
            m_generatedRestoreBarrier2 = m_zk.create(VoltZK.restore_barrier2 + "/counter", null,
                        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (Exception e) {
            VoltDB.crashGlobalVoltDB("Failed to create Zookeeper node: " + e.getMessage(),
                                     false, e);
        }
    }
1338120
def _check_node(self, obj):
        """"""
        if not issubclass(type(obj), self.definition['node_class']):
            raise ValueError("Expected node of class " + self.definition['node_class'].__name__)
        if not hasattr(obj, 'id'):
            raise ValueError("Can't perform operation on unsaved node " + repr(obj))
1290320
static void createIfNotThere(ZooKeeper zookeeper, String znode) throws KeeperException, InterruptedException {
        try {
            create(zookeeper, znode);
        } catch (KeeperException e) {
            if (e.code() != KeeperException.Code.NODEEXISTS) {
                // Rethrow all exceptions, except "node exists",
                // because if the node exists, this method reached its goal.
                throw e;
            }
        }
    }
1010433
public void suspendZooKeeper() {
        val zk = this.zkServer.get();
        Preconditions.checkState(zk != null, "ZooKeeper not started.");

        // Stop, but do not close, the ZK runner.
        zk.stop();
        log.info("ZooKeeper suspended.");
    }
132740
private ZooKeeper createZooKeeper() throws IOException {
    ZooKeeper zk = new ZooKeeper(zkStr, sessionTimeout, wrapWatcher(this));
    for (Map.Entry<String, byte[]> authInfo : authInfos.entries()) {
      zk.addAuthInfo(authInfo.getKey(), authInfo.getValue());
    }
    return zk;
  }
1221545
def recursive_create(path: nil)
      raise ArgumentError, "path is a required argument" if path.nil?

      result = zk.stat(path: path)
      case result.fetch(:rc)
      when Zookeeper::Constants::ZOK
        return
      when Zookeeper::Constants::ZNONODE
        recursive_create(path: File.dirname(path))
        result = zk.create(path: path)

        case result.fetch(:rc)
        when Zookeeper::Constants::ZOK, Zookeeper::Constants::ZNODEEXISTS
          return
        else
          raise Kazoo::Error, "Failed to create node #{path}. Result code: #{result.fetch(:rc)}"
        end
      else
        raise Kazoo::Error, "Failed to create node #{path}. Result code: #{result.fetch(:rc)}"
      end
    end
1244099
public boolean createNode(String path, byte[] value) throws ZooKeeperException {
        return _create(path, value, CreateMode.PERSISTENT);
    }
364877
@Given("^I connect to Zookeeper at '(.+)'$")
    public void connectToZk(String zookeeperHosts) throws InterruptedException {
        commonspec.getZookeeperSecClient().setZookeeperSecConnection(zookeeperHosts, 3000);
        commonspec.getZookeeperSecClient().connectZk();
    }
1184907
@Override
	public ZooClassDef getSchema(Class<?> cls, Node node) {
		ZooClassDef ret = nodeSchemata.get(node).get(cls);
		if (ret == null) {
			if (cls == null) {
				return null;
			}
			//Try virtual/generic schemata
			ret = getSchema(cls.getName());
			if (ret != null) {
				//check (associate also checks compatibility)
				ret.associateJavaTypes(true);
				nodeSchemata.get(node).put(cls, ret);
			}
		}
		return ret;
	}