@Override
public void kill()
{
    try
    {
        if ( quorumPeer != null )
        {
            Field               cnxnFactoryField = QuorumPeer.class.getDeclaredField("cnxnFactory");
            cnxnFactoryField.setAccessible(true);
            ServerCnxnFactory   cnxnFactory = (ServerCnxnFactory)cnxnFactoryField.get(quorumPeer);
            cnxnFactory.closeAll();

            Field               ssField = cnxnFactory.getClass().getDeclaredField("ss");
            ssField.setAccessible(true);
            ServerSocketChannel ss = (ServerSocketChannel)ssField.get(cnxnFactory);
            ss.close();
        }
        close();
    }
    catch ( Exception e )
    {
        e.printStackTrace();
    }
}
 

private void shutdown(QuorumPeer qp) {
  try {
    log.info("Shutting down quorum peer " + qp.getName());
    qp.shutdown();
    Election e = qp.getElectionAlg();
    if (e != null) {
      log.info("Shutting down leader election " + qp.getName());
      e.shutdown();
    } else {
      log.info("No election available to shutdown " + qp.getName());
    }

    log.info("Waiting for " + qp.getName() + " to exit thread");
    qp.join(30000L);
    if (qp.isAlive()) {
      Assert.fail("QP failed to shutdown in 30 seconds: " + qp.getName());
    }
  } catch (InterruptedException var2) {
    log.debug("QP interrupted: " + qp.getName(), var2);
  }

}
 

@Override
public void kill()
{
    try
    {
        if ( quorumPeer != null )
        {
            Field               cnxnFactoryField = QuorumPeer.class.getDeclaredField("cnxnFactory");
            cnxnFactoryField.setAccessible(true);
            ServerCnxnFactory   cnxnFactory = (ServerCnxnFactory)cnxnFactoryField.get(quorumPeer);
            Compatibility.serverCnxnFactoryCloseAll(cnxnFactory);

            Field               ssField = cnxnFactory.getClass().getDeclaredField("ss");
            ssField.setAccessible(true);
            ServerSocketChannel ss = (ServerSocketChannel)ssField.get(cnxnFactory);
            ss.close();
        }
        close();
    }
    catch ( Exception e )
    {
        e.printStackTrace();
    }
}
 

private EmbeddedZookeeper(
    ExecutorService executorService, TemporaryFolder temporaryFolder, Integer replicationFactor)
    throws SaslException {
  this.temporaryFolder = temporaryFolder;
  this.replicationFactor = replicationFactor;
  this.executorService = executorService;
  quorumPeer = QuorumPeer.testingQuorumPeer();
}
 

private void startDistributed() throws IOException {
    logger.info("Starting Embedded ZooKeeper Peer");

    try {
        transactionLog = new FileTxnSnapLog(new File(quorumPeerConfig.getDataLogDir()), new File(quorumPeerConfig.getDataDir()));

        connectionFactory = ServerCnxnFactory.createFactory();
        connectionFactory.configure(quorumPeerConfig.getClientPortAddress(), quorumPeerConfig.getMaxClientCnxns());

        quorumPeer = new QuorumPeer();
        quorumPeer.setClientPortAddress(quorumPeerConfig.getClientPortAddress());
        quorumPeer.setTxnFactory(new FileTxnSnapLog(new File(quorumPeerConfig.getDataLogDir()), new File(quorumPeerConfig.getDataDir())));
        quorumPeer.setQuorumPeers(quorumPeerConfig.getServers());
        quorumPeer.setElectionType(quorumPeerConfig.getElectionAlg());
        quorumPeer.setMyid(quorumPeerConfig.getServerId());
        quorumPeer.setTickTime(quorumPeerConfig.getTickTime());
        quorumPeer.setMinSessionTimeout(quorumPeerConfig.getMinSessionTimeout());
        quorumPeer.setMaxSessionTimeout(quorumPeerConfig.getMaxSessionTimeout());
        quorumPeer.setInitLimit(quorumPeerConfig.getInitLimit());
        quorumPeer.setSyncLimit(quorumPeerConfig.getSyncLimit());
        quorumPeer.setQuorumVerifier(quorumPeerConfig.getQuorumVerifier());
        quorumPeer.setCnxnFactory(connectionFactory);
        quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
        quorumPeer.setLearnerType(quorumPeerConfig.getPeerType());
        quorumPeer.setSyncEnabled(quorumPeerConfig.getSyncEnabled());
        quorumPeer.setQuorumListenOnAllIPs(quorumPeerConfig.getQuorumListenOnAllIPs());

        quorumPeer.start();
    } catch (final IOException ioe) {
        throw new IOException("Failed to start embedded ZooKeeper Peer", ioe);
    } catch (final Exception e) {
        throw new RuntimeException("Failed to start embedded ZooKeeper Peer", e);
    }
}
 

private void initialize() throws IOException {
  // org.apache.zookeeper.test.ClientBase relies on 4lw and the whitelist only contains `srvr`
  // in ZooKeeper 3.5.3 and later (it was less restrictive in previous versions)
  System.setProperty("zookeeper.4lw.commands.whitelist", "*");
  HashMap peers = new HashMap();
  for (int i = 0; i < numNodes; i++) {

    int port = basePort++;
    int portLE = basePort++;

    peers.put(Long.valueOf(i), new QuorumPeer.QuorumServer(
        Long.valueOf(i).longValue(),
        new InetSocketAddress(LOCAL_ADDR, port + 1000),
        new InetSocketAddress(LOCAL_ADDR, portLE + 1000),
        QuorumPeer.LearnerType.PARTICIPANT
    ));
  }

  for (int i = 0; i < numNodes; i++) {

    File dir = Files.createTempDirectory("zk" + i).toFile();

    int portClient = basePort++;
    log.info("creating QuorumPeer " + i + " port " + portClient);
    QuorumPeer s = new QuorumPeer(peers, dir, dir, portClient, 3, i, tickTime, initLimit,
                                  syncLimit
    );
    Assert.assertEquals(portClient, s.getClientPort());

    quorumPeersById.put(i, s);

    if (i == 0) {
      hostPort = LOCAL_ADDR + ":" + portClient;
    } else {
      hostPort = hostPort + "," + LOCAL_ADDR + ":" + portClient;
    }

  }
}
 

public void start() throws IOException {

    JMXEnv.setUp();

    for (int i = 0; i < numNodes; i++) {
      log.info("start QuorumPeer " + i);
      QuorumPeer s = quorumPeersById.get(i);
      s.start();
    }

    log.info("Checking ports " + hostPort);

    for (String hp : hostPort.split(",")) {
      Assert.assertTrue(
          "waiting for server up",
          ClientBase.waitForServerUp(
              hp,
              CONNECTION_TIMEOUT
          )
      );
      log.info(hp + " is accepting client connections");
      try {
        log.info(send4LW(hp, CONNECTION_TIMEOUT, "stat"));
      } catch (TimeoutException | SSLContextException e) {
        log.error(e.getMessage(), e);
      }

    }

    JMXEnv.dump();
    isRunning = true;
  }
 

private QuorumPeer createPeer(ServerCnxnFactory cnxnFactory, QuorumPeerConfig config) throws IOException {
    cnxnFactory.configure(config.getClientPortAddress(),
            config.getMaxClientCnxns());

    QuorumPeer quorumPeer = new QuorumPeer();
    quorumPeer.setClientPortAddress(config.getClientPortAddress());
    quorumPeer.setTxnFactory(new FileTxnSnapLog(
            new File(config.getDataLogDir()),
            new File(config.getDataDir())));
    quorumPeer.setQuorumPeers(config.getServers());
    quorumPeer.setElectionType(config.getElectionAlg());
    quorumPeer.setMyid(config.getServerId());
    quorumPeer.setTickTime(config.getTickTime());
    quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
    quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
    quorumPeer.setInitLimit(config.getInitLimit());
    quorumPeer.setSyncLimit(config.getSyncLimit());
    quorumPeer.setQuorumVerifier(config.getQuorumVerifier());
    quorumPeer.setCnxnFactory(cnxnFactory);
    quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
    quorumPeer.setLearnerType(config.getPeerType());
    quorumPeer.setSyncEnabled(config.getSyncEnabled());
    quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());

    quorumPeer.start();

    return quorumPeer;
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testSegmentation3() throws Exception {
    sesTimeout = 5000;

    Ignite node0 = startGrid(0);

    final CountDownLatch l = new CountDownLatch(1);

    node0.events().localListen(new IgnitePredicate<Event>() {
        @Override public boolean apply(Event evt) {
            l.countDown();

            return false;
        }
    }, EventType.EVT_NODE_SEGMENTED);

    List<TestingZooKeeperServer> srvs = zkCluster.getServers();

    assertEquals(3, srvs.size());

    try {
        srvs.get(0).stop();
        srvs.get(1).stop();

        QuorumPeer qp = srvs.get(2).getQuorumPeer();

        // Zookeeper's socket timeout [tickTime * initLimit] + 5 additional seconds for other logic
        assertTrue(l.await(qp.getTickTime() * qp.getInitLimit() + 5000, TimeUnit.MILLISECONDS));
    }
    finally {
        zkCluster.close();

        zkCluster = ZookeeperDiscoverySpiTestUtil.createTestingCluster(ZK_SRVS);

        zkCluster.start();
    }
}
 

public QuorumPeer getQuorumPeer()
{
    if (isCluster()) {
        return ((TestingQuorumPeerMain) main).getTestingQuorumPeer();
    }
    throw new UnsupportedOperationException();
}
 

@VisibleForTesting
public static String configToConnectionString(QuorumVerifier data) throws Exception
{
    StringBuilder sb = new StringBuilder();
    for ( QuorumPeer.QuorumServer server : data.getAllMembers().values() )
    {
        if ( server.clientAddr == null )
        {
            // Invalid client address configuration in zoo.cfg
            continue;
        }
        if ( sb.length() != 0 )
        {
            sb.append(",");
        }
        String hostAddress;
        if ( server.clientAddr.getAddress().isAnyLocalAddress() )
        {
            hostAddress = Compatibility.getHostAddress(server);
        }
        else
        {
            hostAddress = server.clientAddr.getAddress().getHostAddress();
        }
        sb.append(hostAddress).append(":").append(server.clientAddr.getPort());
    }

    return sb.toString();
}
 

private void assertConfig(QuorumVerifier config, Collection<InstanceSpec> instances)
{
    for ( InstanceSpec instance : instances )
    {
        QuorumPeer.QuorumServer quorumServer = config.getAllMembers().get((long)instance.getServerId());
        Assert.assertNotNull(quorumServer, String.format("Looking for %s - found %s", instance.getServerId(), config.getAllMembers()));
        Assert.assertEquals(quorumServer.clientAddr.getPort(), instance.getPort());
    }
}
 

private void startDistributed() throws IOException {
    logger.info("Starting Embedded ZooKeeper Peer");

    try {
        transactionLog = new FileTxnSnapLog(quorumPeerConfig.getDataLogDir(), quorumPeerConfig.getDataDir());

        connectionFactory = ServerCnxnFactory.createFactory();
        connectionFactory.configure(quorumPeerConfig.getClientPortAddress(), quorumPeerConfig.getMaxClientCnxns());

        quorumPeer = new QuorumPeer();
        quorumPeer.setTxnFactory(new FileTxnSnapLog(quorumPeerConfig.getDataLogDir(), quorumPeerConfig.getDataDir()));
        quorumPeer.setElectionType(quorumPeerConfig.getElectionAlg());
        quorumPeer.setMyid(quorumPeerConfig.getServerId());
        quorumPeer.setTickTime(quorumPeerConfig.getTickTime());
        quorumPeer.setMinSessionTimeout(quorumPeerConfig.getMinSessionTimeout());
        quorumPeer.setMaxSessionTimeout(quorumPeerConfig.getMaxSessionTimeout());
        quorumPeer.setInitLimit(quorumPeerConfig.getInitLimit());
        quorumPeer.setSyncLimit(quorumPeerConfig.getSyncLimit());
        quorumPeer.setQuorumVerifier(quorumPeerConfig.getQuorumVerifier(), false);
        quorumPeer.setCnxnFactory(connectionFactory);
        quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
        quorumPeer.setLearnerType(quorumPeerConfig.getPeerType());
        quorumPeer.setSyncEnabled(quorumPeerConfig.getSyncEnabled());
        quorumPeer.setQuorumListenOnAllIPs(quorumPeerConfig.getQuorumListenOnAllIPs());

        quorumPeer.start();
    } catch (final IOException ioe) {
        throw new IOException("Failed to start embedded ZooKeeper Peer", ioe);
    } catch (final Exception e) {
        throw new RuntimeException("Failed to start embedded ZooKeeper Peer", e);
    }
}
 

public SpliceZoo(QuorumPeerConfig config, int number) throws IOException {
	this.config = config;
	try {
		if (QuorumPeer.class.getMethod("testingQuorumPeer", null) != null)
			this.peer = (QuorumPeer) QuorumPeer.class.getMethod("testingQuorumPeer", null).invoke(null,null);
		else
			this.peer = QuorumPeer.class.newInstance();
	} catch (Exception e) {
		throw new RuntimeException("Quorum Peer Signature Issue for Unit Tests");
	}
	ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
	cnxnFactory.configure(config.getClientPortAddress(),config.getMaxClientCnxns());
	peer.setClientPortAddress(config.getClientPortAddress());
	peer.setTxnFactory(new FileTxnSnapLog(new File(config.getDataLogDir()),
                    new File(config.getDataDir())));
	peer.setQuorumPeers(config.getServers());
	peer.setElectionType(config.getElectionAlg());
	peer.setMyid(config.getServerId());
	peer.setTickTime(config.getTickTime());
	peer.setMinSessionTimeout(config.getMinSessionTimeout());
	peer.setMaxSessionTimeout(config.getMaxSessionTimeout());
	peer.setInitLimit(config.getInitLimit());
	peer.setSyncLimit(config.getSyncLimit());
	peer.setQuorumVerifier(config.getQuorumVerifier());
	peer.setCnxnFactory(cnxnFactory);
	peer.setZKDatabase(new ZKDatabase(peer.getTxnFactory()));
	peer.setLearnerType(config.getPeerType());
	peer.setMyid(number);
}
 

public QuorumPeer getQuorumPeer()
{
    return main.getQuorumPeer();
}
 

@Override
public QuorumPeer getQuorumPeer()
{
    throw new UnsupportedOperationException();
}
 

@Override
public QuorumPeer getQuorumPeer()
{
    return quorumPeer;
}
 

public Map<Long, QuorumPeer.QuorumServer> getServers() {
  return zkProps.getServers();
}
 

public QuorumPeer getTestingQuorumPeer()
{
    return quorumPeer;
}
 

public QuorumPeer getQuorumPeer();