Runs a ZooKeeper {@link QuorumPeer} if further peers are configured or a single
{@link ZooKeeperServer} if no further peers are configured.

@param zkConfigFile ZooKeeper config file 'zoo.cfg'
@param peerId       ID for the 'myid' file
969470
public static void runFlinkZkQuorumPeer(String zkConfigFile, int peerId) throws Exception {

		Properties zkProps = new Properties();

		try (InputStream inStream = new FileInputStream(new File(zkConfigFile))) {
			zkProps.load(inStream);
		}

		LOG.info("Configuration: " + zkProps);

		// Set defaults for required properties
		setRequiredProperties(zkProps);

		// Write peer id to myid file
		writeMyIdToDataDir(zkProps, peerId);

		// The myid file needs to be written before creating the instance. Otherwise, this
		// will fail.
		QuorumPeerConfig conf = new QuorumPeerConfig();
		conf.parseProperties(zkProps);

		if (conf.isDistributed()) {
			// Run quorum peer
			LOG.info("Running distributed ZooKeeper quorum peer (total peers: {}).",
					conf.getServers().size());

			QuorumPeerMain qp = new QuorumPeerMain();
			qp.runFromConfig(conf);
		}
		else {
			// Run standalone
			LOG.info("Running standalone ZooKeeper quorum peer.");

			ZooKeeperServerMain zk = new ZooKeeperServerMain();
			ServerConfig sc = new ServerConfig();
			sc.readFrom(conf);
			zk.runFromConfig(sc);
		}
	}
1121407
@Override
    public SimpleZooKeeperClient doConnect() throws IOException, InterruptedException {
        LOG.info("Connecting to zookeeper server - " + zooConf.getOnlineServerList());
        return new SimpleZooKeeperClient(zooConf.getOnlineServerList(),
                zooConf.getOnlineDigestAuth(), new ServerServiceWatcher(),
                zooConf.getSessionTimeout(), zooConf.getConnTimeout());
    }
654424
@Override
  public void join(String peer) throws Exception {
    LOG.debug("Follower joins in.");
    try {
      LOG.debug("Query leader from {}", peer);
      Message query = MessageBuilder.buildQueryLeader();
      MessageTuple tuple = null;
      while (true) {
        // The joiner might gets started before the cluster gets started.
        // Instead of reporting failure immediately, the follower will retry
        // joining the cluster until joining the cluster successfully.
        try {
          sendMessage(peer, query);
          tuple = filter.getExpectedMessage(MessageType.QUERY_LEADER_REPLY,
                                            peer, config.getTimeoutMs());
          break;
        } catch (TimeoutException ex) {
          long retryInterval = 1;
          LOG.warn("Timeout while contacting leader, going to retry after {} "
              + "second.", retryInterval);
          // Waits for 1 second.
          Thread.sleep(retryInterval * 1000);
        }
      }
      this.electedLeader = tuple.getMessage().getReply().getLeader();
      LOG.debug("Got current leader {}", this.electedLeader);

      /* -- Synchronizing phase -- */
      while (true) {
        try {
          joinSynchronization();
          break;
        } catch (TimeoutException | BackToElectionException ex) {
          LOG.debug("Timeout({} ms) in synchronizing, retrying. Last zxid : {}",
                    getSyncTimeoutMs(),
                    persistence.getLatestZxid());
          transport.clear(electedLeader);
          clearMessageQueue();
        }
      }
      // See if it can be restored from the snapshot file.
      restoreFromSnapshot();
      // Delivers all transactions in log before entering broadcasting phase.
      deliverUndeliveredTxns();

      /* -- Broadcasting phase -- */
      // Initialize the vote for leader election.
      this.election.specifyLeader(this.electedLeader);
      changePhase(Phase.BROADCASTING);
      accepting();
    } catch (InterruptedException e) {
      LOG.debug("Participant is canceled by user.");
      throw e;
    } catch (TimeoutException e) {
      LOG.debug("Didn't hear message from {} for {} milliseconds. Going"
                + " back to leader election.",
                this.electedLeader,
                this.config.getTimeoutMs());
      if (persistence.getLastSeenConfig() == null) {
        throw new JoinFailure("Fails to join cluster.");
      }
    } catch (BackToElectionException e) {
      LOG.debug("Got GO_BACK message from queue, going back to electing.");
      if (persistence.getLastSeenConfig() == null) {
        throw new JoinFailure("Fails to join cluster.");
      }
    } catch (LeftCluster e) {
      LOG.debug("Exit running : {}", e.getMessage());
      throw e;
    } catch (Exception e) {
      LOG.error("Caught exception", e);
      throw e;
    } finally {
      changePhase(Phase.FINALIZING);
    }
  }
849135
public static CuratorFramework newRootFluoCurator(FluoConfiguration config) {
    return newCurator(ZookeeperUtil.parseServers(config.getInstanceZookeepers()),
        config.getZookeeperTimeout(), config.getZookeeperSecret());
  }
248610
public CuratorFramework mkClient(Map conf, List<String> servers, Object port,
                                     String root, final WatcherCallBack watcher) {

        CuratorFramework fk = Utils.newCurator(conf, servers, port, root);

        fk.getCuratorListenable().addListener(new CuratorListener() {
            @Override
            public void eventReceived(CuratorFramework _fk, CuratorEvent e) throws Exception {
                if (e.getType().equals(CuratorEventType.WATCHED)) {
                    WatchedEvent event = e.getWatchedEvent();

                    watcher.execute(event.getState(), event.getType(), event.getPath());
                }

            }
        });

        fk.getUnhandledErrorListenable().addListener(new UnhandledErrorListener() {
            @Override
            public void unhandledError(String msg, Throwable error) {
                String errmsg = "Unrecoverable zookeeper error, halting process: " + msg;
                LOG.error(errmsg, error);
                JStormUtils.halt_process(1, "Unrecoverable zookeeper error");

            }
        });
        fk.start();
        return fk;
    }
449246
void beginSynchronizing() throws IOException {
    // Synchronization is performed in other threads.
    Zxid lastZxid = persistence.getLatestZxid();
    ClusterConfiguration clusterConfig = persistence.getLastSeenConfig();
    long proposedEpoch = persistence.getProposedEpoch();
    for (PeerHandler ph : this.quorumMap.values()) {
      ph.setSyncTask(new SyncPeerTask(ph.getServerId(), ph.getLastZxid(),
                                      lastZxid, clusterConfig),
                     proposedEpoch);
      ph.startSynchronizingTask();
    }
  }
1136149
def setup_zk
      @zk = ZK.new(@zkservers) if @zkservers
      @zk.register(redis_nodes_path) { |event| handle_zk_event(event) }
      if @safe_mode
        @zk.on_expired_session { purge_clients }
      end
      @zk.on_connected { @zk.stat(redis_nodes_path, :watch => true) }
      @zk.stat(redis_nodes_path, :watch => true)
      update_znode_timestamp
    end
1275450
private static TestingServer startLocalZooKeeper() throws Exception {
        // ZooKeeper is too noisy by default.
        ((Logger) LoggerFactory.getLogger("org.apache.zookeeper")).setLevel(Level.ERROR);

        // Start the testing server.
        TestingServer zooKeeperServer = new TestingServer(2181);

        // Configure EmoDB to use the testing server.
        System.setProperty("dw.zooKeeper.connectString", zooKeeperServer.getConnectString());

        return zooKeeperServer;
    }
1368255
private static void registerDnProtocolAddress(AvatarZooKeeperClient zk,
      Configuration originalConf, Configuration conf, boolean toOverwrite)
      throws UnsupportedEncodingException, IOException {
    LOG.info("Updating Service Address information in ZooKeeper");
    registerSocketAddress(zk,
        originalConf.get(NameNode.DATANODE_PROTOCOL_ADDRESS),
        conf.get(NameNode.DATANODE_PROTOCOL_ADDRESS), toOverwrite);
    registerAliases(zk, conf, FSConstants.DFS_NAMENODE_DN_ALIASES,
        conf.get(NameNode.DATANODE_PROTOCOL_ADDRESS), toOverwrite);
  }
426194
@Override
	public Protos.TaskID newTaskID() throws Exception {
		synchronized (startStopLock) {
			verifyIsRunning();

			int nextCount;
			boolean success;
			do {
				ZooKeeperVersionedValue<Integer> count = totalTaskCountInZooKeeper.getVersionedValue();
				nextCount = count.getValue() + 1;
				success = totalTaskCountInZooKeeper.trySetCount(count, nextCount);
			}
			while (!success);

			Protos.TaskID taskID = Protos.TaskID.newBuilder().setValue(TASKID_FORMAT.format(nextCount)).build();
			return taskID;
		}
	}
889311
private boolean zkPathExists(String path) throws IOException {
    try {
      return zk.exists(path, false) != null;
    } catch (KeeperException e) {
      keeperException("Unrecoverable ZooKeeper error checking if " +
          path + " exists", e);
    } catch (InterruptedException e) {
      interruptedException("Interrupted checking if ZooKeeper path " +
          path + " exists", e);
    }
    return false; // Should never be reached
  }
82657
@Singleton
  @Provides
  private YarnTwillRunnerService provideYarnTwillRunnerService(CConfiguration configuration,
                                                               YarnConfiguration yarnConfiguration,
                                                               LocationFactory locationFactory) {
    String zkConnectStr = configuration.get(Constants.Zookeeper.QUORUM) +
                          configuration.get(Constants.CFG_TWILL_ZK_NAMESPACE);

    // Copy the yarn config and set the max heap ratio.
    YarnConfiguration yarnConfig = new YarnConfiguration(yarnConfiguration);
    yarnConfig.set(Constants.CFG_TWILL_RESERVED_MEMORY_MB, configuration.get(Constants.CFG_TWILL_RESERVED_MEMORY_MB));
    YarnTwillRunnerService runner = new YarnTwillRunnerService(yarnConfig,
                                                               zkConnectStr,
                                                               LocationFactories.namespace(locationFactory, "twill"));

    // Set JVM options based on configuration
    runner.setJVMOptions(configuration.get(Constants.Container.PROGRAM_JVM_OPTS));

    return runner;
  }
433702
void waitProposedEpochFromQuorum()
      throws InterruptedException, TimeoutException, IOException {
    ClusterConfiguration currentConfig = persistence.getLastSeenConfig();
    long acknowledgedEpoch = persistence.getAckEpoch();
    // Waits PROPOED_EPOCH from a quorum of peers in current configuraion.
    while (this.quorumMap.size() < getQuorumSize() - 1) {
      MessageTuple tuple = filter.getExpectedMessage(MessageType.PROPOSED_EPOCH,
                                                     null,
                                                     config.getTimeoutMs());
      Message msg = tuple.getMessage();
      String source = tuple.getServerId();
      ZabMessage.ProposedEpoch epoch = msg.getProposedEpoch();
      ClusterConfiguration peerConfig =
        ClusterConfiguration.fromProto(epoch.getConfig(), source);
      long peerProposedEpoch = epoch.getProposedEpoch();
      long peerAckedEpoch = epoch.getCurrentEpoch();
      int syncTimeoutMs = epoch.getSyncTimeout();
      Zxid peerVersion = peerConfig.getVersion();
      Zxid selfVersion = currentConfig.getVersion();
      // If the peer's config version doesn't match leader's config version,
      // we'll check if the peer has the more likely "correct" configuration.
      if (!peerVersion.equals(selfVersion)) {
        LOG.debug("{}'s config version {} is different with leader's {}",
                  peerVersion, selfVersion);
        if (peerAckedEpoch > acknowledgedEpoch ||
            (peerAckedEpoch == acknowledgedEpoch &&
             peerVersion.compareTo(selfVersion) > 0)) {
          LOG.debug("{} probably has right configuration, go back to "
              + "leader election.",
              source);
          // TODO : current we just go back to leader election, probably we want
          // to select the peer as leader.
          throw new BackToElectionException();
        }
      }
      // Rejects peer who is not in the current configuration.
      if (!currentConfig.contains(source)) {
        LOG.debug("Current configuration doesn't contain {}, ignores it.",
                  source);
        continue;
      }
      if (this.quorumMap.containsKey(source)) {
        throw new RuntimeException("Quorum set has already contained "
            + source + ", probably a bug?");
      }
      LOG.debug("Got PROPOSED_EPOCH from {}", source);
      PeerHandler ph = new PeerHandler(source, transport,
                                       config.getTimeoutMs()/3);
      ph.setLastProposedEpoch(peerProposedEpoch);
      ph.setSyncTimeoutMs(syncTimeoutMs);
      this.quorumMap.put(source, ph);
    }
    LOG.debug("Got proposed epoch from a quorum.");
  }
1079232
private void joinCluster() throws InterruptedException, IOException {
        while (true) {
            NodeInfo myInfo;
            try {
                myInfo = new NodeInfo(NodeState.Fresh.toString(), zk.get().getSessionId());
            } catch (ZooKeeperConnectionException e) {
                throw ZKException.from(e);
            }
            byte[] encoded = JsonUtil.asJSONBytes(myInfo);
            if (ZKUtils.createEphemeral(zk, "/" + name + "/nodes/" + myNodeID, encoded)) {
                return;
            } else {
                Stat stat = new Stat();
                try {
                    byte[] bytes = zk.get().getData("/" + name + "/nodes/" + myNodeID, false, stat);
                    NodeInfo nodeInfo = JsonUtil.fromJSON(bytes, NodeInfo.class);
                    if(nodeInfo.connectionID == zk.get().getSessionId()) {
                        // As it turns out, our session is already registered!
                        return;
                    }
                } catch (ZooKeeperConnectionException e) {
                    throw ZKException.from(e);
                } catch (KeeperException e) {
                    throw ZKException.from(e);
                }
            }
            LOG.warn("Unable to register with Zookeeper on launch. " +
                    "Is {} already running on this host? Retrying in 1 second...", name);
            Thread.sleep(1000);
        }
    }
254275
public void setPeerId(ByteBuffer peerId) {
    if (peerId != null) {
      this.peerId = peerId;
      this.hexPeerId = TorrentUtils.byteArrayToHexString(peerId.array());
    } else {
      this.peerId = null;
      this.hexPeerId = null;
    }
  }
901693
public static void createPersistentZKNodes(ZooKeeper zk) {
        LinkedList<ZKUtil.StringCallback> callbacks = new LinkedList<ZKUtil.StringCallback>();
        for (int i=0; i < VoltZK.ZK_HIERARCHY.length; i++) {
            ZKUtil.StringCallback cb = new ZKUtil.StringCallback();
            callbacks.add(cb);
            zk.create(VoltZK.ZK_HIERARCHY[i], null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, cb, null);
        }
        for (ZKUtil.StringCallback cb : callbacks) {
            try {
                cb.get();
            } catch (org.apache.zookeeper_voltpatches.KeeperException.NodeExistsException e) {
                // this is an expected race.
            } catch (Exception e) {
                VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
            }
        }
    }
395763
private void reconnectZooKeeper() throws InterruptedException, IOException {
    LOG.info("Try to reconnect ZooKeeper " + zkQuorum);
    
    if (zooKeeper != null) {
      zooKeeper.close();
    }
    connectZooKeeper();
  }