Java 类org.apache.zookeeper.data.Stat 实例源码

/**
 * Create a new Stat, fill in dummy values trying to catch Assert.failure
 * to copy in client or server code.
 *
 * @return a new stat with dummy values
 */
private Stat newStat() {
    Stat stat = new Stat();

    stat.setAversion(100);
    stat.setCtime(100);
    stat.setCversion(100);
    stat.setCzxid(100);
    stat.setDataLength(100);
    stat.setEphemeralOwner(100);
    stat.setMtime(100);
    stat.setMzxid(100);
    stat.setNumChildren(100);
    stat.setPzxid(100);
    stat.setVersion(100);

    return stat;
}

private void connect(TxZookeeperConfig config) {
    try {
        zooKeeper = new ZooKeeper(config.getHost(), config.getSessionTimeOut(), watchedEvent -> {
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                // 放开闸门, wait在connect方法上的线程将被唤醒
                COUNT_DOWN_LATCH.countDown();
            }
        });
        COUNT_DOWN_LATCH.await();
        Stat stat = zooKeeper.exists(rootPath, false);
        if (stat == null) {
            zooKeeper.create(rootPath, rootPath.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    } catch (Exception e) {
        throw new TransactionIoException(e);
    }


}

public void processResult(int rc, String path,
        Object ctx, byte[] data, Stat stat) {
    if (rc == KeeperException.Code.NONODE.intValue()) {
        // we can just ignore because the child watcher takes care of this
        return;
    }
    if (rc != KeeperException.Code.OK.intValue()) {
        zk.getData(myNode, (Watcher)ctx, this, ctx);
    }
    int currVer = stat.getVersion();
    if (currVer != lastVer) {
        String parts[] = new String(data).split(" ", 2);
        myInstance.configure(parts[1]);
        lastVer = currVer;
    }
}

@Test
public void testCreateEphemeralZNode() 
throws KeeperException, InterruptedException, IOException {
    ZSession session = createSession("30");

    WebResource wr = znodesr.path("/")
        .queryParam("op", "create")
        .queryParam("name", "ephemeral-test")
        .queryParam("ephemeral", "true")
        .queryParam("session", session.id)
        .queryParam("null", "true");

    Builder b = wr.accept(MediaType.APPLICATION_JSON);
    ClientResponse cr = b.post(ClientResponse.class);
    Assert.assertEquals(ClientResponse.Status.CREATED, cr.getClientResponseStatus());

    Stat stat = new Stat();
    zk.getData("/ephemeral-test", false, stat);

    ZooKeeper sessionZK = ZooKeeperService.getClient(CONTEXT_PATH, session.id);
    Assert.assertEquals(stat.getEphemeralOwner(), sessionZK.getSessionId());
}

public void testRoundTrip() throws IOException {
    MultiResponse response = new MultiResponse();

    response.add(new OpResult.CheckResult());
    response.add(new OpResult.CreateResult("foo-bar"));
    response.add(new OpResult.DeleteResult());

    Stat s = new Stat();
    s.setCzxid(546);
    response.add(new OpResult.SetDataResult(s));

    MultiResponse decodedResponse = codeDecode(response);

    Assert.assertEquals(response, decodedResponse);
    Assert.assertEquals(response.hashCode(), decodedResponse.hashCode());
}

/**
 * The asynchronous version of getACL.
 *
 * @see #getACL(String, Stat)
 */
public void getACL(final String path, Stat stat, ACLCallback cb,
        Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getACL);
    GetACLRequest request = new GetACLRequest();
    request.setPath(serverPath);
    GetACLResponse response = new GetACLResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, null);
}

public static void main(String[] args) {
    String connStr = "127.0.0.1:2181";
    try {
        ZooKeeper zookeeper = new ZooKeeper(connStr, 100000,
            new Watcher() {
                public void process(WatchedEvent event) {
                    logger.debug("监控被触发的事件");
                }
            });
        Stat stat = new Stat();
        byte[] result = zookeeper.getData("/saf_service/com.ipd.testjsf.HelloBaontService/providers", false, stat);
        zookeeper.setData("/saf_service/com.ipd.testjsf.HelloBaontService/providers", result, stat.getVersion());
        System.out.println(result);
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

/**
 * Creates the specified node with the specified data and watches it.
 *
 * <p>Throws an exception if the node already exists.
 *
 * <p>The node created is persistent and open access.
 *
 * <p>Returns the version number of the created node if successful.
 *
 * @param zkw zk reference
 * @param znode path of node to create
 * @param data data of node to create
 * @return version of node created
 * @throws KeeperException if unexpected zookeeper exception
 * @throws KeeperException.NodeExistsException if node already exists
 */
public static int createAndWatch(ZooKeeperWatcher zkw,
    String znode, byte [] data)
throws KeeperException, KeeperException.NodeExistsException {
  try {
    zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
        CreateMode.PERSISTENT);
    Stat stat = zkw.getRecoverableZooKeeper().exists(znode, zkw);
    if (stat == null){
      // Likely a race condition. Someone deleted the znode.
      throw KeeperException.create(KeeperException.Code.SYSTEMERROR,
          "ZK.exists returned null (i.e.: znode does not exist) for znode=" + znode);
    }
   return stat.getVersion();
  } catch (InterruptedException e) {
    zkw.interruptedException(e);
    return -1;
  }
}

@Test(timeout = 60000)
public void testRootWatchTriggered() throws Exception {
    class MyWatcher implements Watcher{
        boolean fired=false;
        public void process(WatchedEvent event) {
            if(event.getPath().equals("/"))
                fired=true;
        }
    }
    MyWatcher watcher=new MyWatcher();
    // set a watch on the root node
    dt.getChildren("/", new Stat(), watcher);
    // add a new node, should trigger a watch
    dt.createNode("/xyz", new byte[0], null, 0, dt.getNode("/").stat.getCversion()+1, 1, 1);
    Assert.assertFalse("Root node watch not triggered",!watcher.fired);
}

public boolean deleteNode(String nodePath) {
    if (connected) {
        try {
            Stat s = zooKeeper.exists(nodePath, false);
            if (s != null) {
                List<String> children = zooKeeper.getChildren(nodePath,
                        false);
                for (String child : children) {
                    String node = nodePath + "/" + child;
                    deleteNode(node);
                }
                zooKeeper.delete(nodePath, -1);
            }
            return true;
        } catch (Exception e) {
            LoggerFactory.getLogger().error(
                    "Error occurred deleting node: " + nodePath, e);
        }
    }
    return false;
}

/**
 * verify that if create znode results in nodeexists and that znode is deleted
 * before exists() watch is set then the return of the exists() method results
 * in attempt to re-create the znode and become active
 */
@Test
public void testCreateNodeResultRetryNoNode() {
  elector.joinElection(data);

  elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  verifyExistCall(1);

  elector.processResult(Code.NONODE.intValue(), ZK_LOCK_NAME, mockZK,
      (Stat) null);
  Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();
  Mockito.verify(mockZK, Mockito.times(4)).create(ZK_LOCK_NAME, data,
      Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
}

private void connect(TccZookeeperConfig config) {
    try {
        zooKeeper = new ZooKeeper(config.getHost(), config.getSessionTimeOut(), watchedEvent -> {
            if (watchedEvent.getState() == Watcher.Event.KeeperState.SyncConnected) {
                // 放开闸门, wait在connect方法上的线程将被唤醒
                LATCH.countDown();
            }
        });
        LATCH.await();
        Stat stat = zooKeeper.exists(rootPathPrefix, false);
        if (stat == null) {
            zooKeeper.create(rootPathPrefix, rootPathPrefix.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    } catch (Exception e) {
        throw new TccRuntimeException(e);
    }


}

@Test
public void testDefaultConnection() throws Exception {
  // Default root from sabot-module.conf
  assertNull(zooKeeperServer.getZKClient().exists("/dremio/test-path", false));

  final SabotConfig config = DEFAULT_SABOT_CONFIG
      .withValue(ZK_ROOT, ConfigValueFactory.fromAnyRef("dremio/test-path"))
      .withValue(CLUSTER_ID, ConfigValueFactory.fromAnyRef("test-cluster-id"));

  try(ZKClusterClient client = new ZKClusterClient(config, new Provider<Integer>() {
    @Override
    public Integer get() {
      return zooKeeperServer.getPort();
    }
  })) {
    client.start();
    ZKServiceSet serviceSet = client.newServiceSet("coordinator");
    serviceSet.register(NodeEndpoint.newBuilder().setAddress("foo").build());


    Stat stat = zooKeeperServer.getZKClient().exists("/dremio/test-path/test-cluster-id/coordinator", false);
    assertNotNull(stat);
    assertEquals(1, stat.getNumChildren());
  }
}

/**
 * Gets the status of a node in ZooKeeper
 *
 * @param p_path
 *     the node path
 * @param p_watcher
 *     the watcher
 * @return true if the node exists, fals eotherwise
 * @throws ZooKeeperException
 *     if ZooKeeper could not accessed
 */
public Stat getStatus(final String p_path, final Watcher p_watcher) throws ZooKeeperException {
    Stat ret;

    assert p_path != null;

    try {
        if (m_zookeeper == null) {
            connect();
        }

        if (!p_path.isEmpty()) {
            ret = m_zookeeper.exists(m_path + '/' + p_path, p_watcher);
        } else {
            ret = m_zookeeper.exists(m_path, p_watcher);
        }
    } catch (final KeeperException | InterruptedException e) {
        throw new ZooKeeperException("Could not access ZooKeeper", e);
    }

    return ret;
}

@Test
public void testCreateEphemeralZNode() 
throws KeeperException, InterruptedException, IOException {
    ZSession session = createSession("30");

    WebResource wr = znodesr.path("/")
        .queryParam("op", "create")
        .queryParam("name", "ephemeral-test")
        .queryParam("ephemeral", "true")
        .queryParam("session", session.id)
        .queryParam("null", "true");

    Builder b = wr.accept(MediaType.APPLICATION_JSON);
    ClientResponse cr = b.post(ClientResponse.class);
    assertEquals(ClientResponse.Status.CREATED, cr.getClientResponseStatus());

    Stat stat = new Stat();
    zk.getData("/ephemeral-test", false, stat);

    ZooKeeper sessionZK = ZooKeeperService.getClient(CONTEXT_PATH, session.id);
    assertEquals(stat.getEphemeralOwner(), sessionZK.getSessionId());
}

public boolean createNode(String parent, String nodeName) {
    if (connected) {
        try {
            String[] nodeElements = nodeName.split("/");
            for (String nodeElement : nodeElements) {
                String node = parent + "/" + nodeElement;
                Stat s = zooKeeper.exists(node, false);
                if (s == null) {
                    zooKeeper.create(node, this.encryptionManager
                            .encryptData(null), Ids.OPEN_ACL_UNSAFE,
                            CreateMode.PERSISTENT);
                    parent = node;
                }
            }
            return true;
        } catch (Exception e) {
            LoggerFactory.getLogger().error(
                    "Error occurred creating node: " + parent + "/"
                            + nodeName, e);
        }
    }
    return false;
}

@Override
public CuratorFramework get() {
    String quorum = zookeeperConfig.getQuorum();
    String statsPath = zookeeperConfig.getPropertyServicePath();
    String connectionString = quorum + (statsPath == null ? "" : statsPath);

    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    LOGGER.info("Initiating Curator connection to Zookeeper using: [{}]", connectionString);
    // Use chroot so all subsequent paths are below /stroom-stats to avoid conflicts with hbase/zookeeper/kafka etc.
    CuratorFramework client = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
    client.start();

    try {
        // Ensure the chrooted root path exists (i.e. /stroom-stats)
        Stat stat = client.checkExists().forPath("/");
        if (stat == null) {
            LOGGER.info("Creating chroot-ed root node inside " + statsPath);
            client.create().forPath("/");
        }
    } catch (Exception e) {
        throw new RuntimeException("Error connecting to zookeeper using connection String: " + connectionString, e);
    }
    return client;
}

@Test
public void testSetDataVersionMismatchInLoop() throws Exception {
  String znode = "/hbase/splitWAL/9af7cfc9b15910a0b3d714bf40a3248f";
  Configuration conf = TEST_UTIL.getConfiguration();
  ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "testSetDataVersionMismatchInLoop",
      abortable, true);
  String ensemble = ZKConfig.getZKQuorumServersString(conf);
  RecoverableZooKeeper rzk = ZKUtil.connect(conf, ensemble, zkw);
  rzk.create(znode, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  rzk.setData(znode, "OPENING".getBytes(), 0);
  Field zkField = RecoverableZooKeeper.class.getDeclaredField("zk");
  zkField.setAccessible(true);
  int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
  ZookeeperStub zkStub = new ZookeeperStub(ensemble, timeout, zkw);
  zkStub.setThrowExceptionInNumOperations(1);
  zkField.set(rzk, zkStub);
  byte[] opened = "OPENED".getBytes();
  rzk.setData(znode, opened, 1);
  byte[] data = rzk.getData(znode, false, new Stat());
  assertTrue(Bytes.equals(opened, data));
}

@Override
public Stat setData(String path, byte[] data, int version)
        throws KeeperException, InterruptedException {
    int count = 0;
    do {
        try {
            return super.setData(path, data, version);
        } catch (KeeperException.ConnectionLossException e) {
            LoggerFactory.getLogger().warn(
                    "ZooKeeper connection lost.  Trying to reconnect.");
            Stat s = exists(path, false);
            if (s != null) {
                if (getData(path, false, s) == data) {
                    return s;
                }
            } else {
                return null;
            }
        }
    } while (!closed && (limit == -1 || count++ < limit));
    return null;
}

/**
 * 更新 List<Participant>  只更新这一个字段数据
 *
 * @param mythTransaction 实体对象
 */
@Override
public int updateParticipant(MythTransaction mythTransaction) throws MythRuntimeException {

    final String path =
            RepositoryPathUtils
                    .buildZookeeperRootPath(rootPathPrefix, mythTransaction.getTransId());
    try {
        byte[] content = zooKeeper.getData(path,
                false, new Stat());
        final CoordinatorRepositoryAdapter adapter =
                objectSerializer.deSerialize(content, CoordinatorRepositoryAdapter.class);

        adapter.setContents(objectSerializer.serialize(mythTransaction.getMythParticipants()));
        zooKeeper.create(path,
                objectSerializer.serialize(adapter),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        return CommonConstant.SUCCESS;
    } catch (Exception e) {
        throw new MythRuntimeException(e);
    }

}

public Stat setData(String path, byte data[], int version, long zxid,
        long time) throws KeeperException.NoNodeException {
    Stat s = new Stat();
    DataNode n = nodes.get(path);
    if (n == null) {
        throw new KeeperException.NoNodeException();
    }
    byte lastdata[] = null;
    synchronized (n) {
        lastdata = n.data;
        n.data = data;
        n.stat.setMtime(time);
        n.stat.setMzxid(zxid);
        n.stat.setVersion(version);
        n.copyStat(s);
    }
    // now update if the path is in a quota subtree.
    String lastPrefix = getMaxPrefixWithQuota(path);
    if(lastPrefix != null) {
      this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
          - (lastdata == null ? 0 : lastdata.length));
    }
    dataWatches.triggerWatch(path, EventType.NodeDataChanged);
    return s;
}

/**
 * The asynchronous version of getACL.
 *
 * @see #getACL(String, Stat)
 */
public void getACL(final String path, Stat stat, ACLCallback cb,
        Object ctx)
{
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getACL);
    GetACLRequest request = new GetACLRequest();
    request.setPath(serverPath);
    GetACLResponse response = new GetACLResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb,
            clientPath, serverPath, ctx, null);
}

public void process(WatchedEvent event) {
    if (!closed) {
        try {
            if (event.getType() != EventType.NodeDeleted) {

                Stat s = zooKeeper.exists(nodePath, this);
                if (s != null) {
                    zookeeper.getChildren(nodePath, this);
                }
            }
        } catch (Exception e) {
            LoggerFactory.getLogger().error(
                    "Error occured re-adding node watcherfor node "
                            + nodePath, e);
        }
        nodeListener.processEvent(event.getPath(), event.getType()
                .name(), null);
    }
}

@Override
protected void modelModifiedExternally() {
    ZnodeModel znodeModel = getModel();

    if (znodeModel.isDestroyed()) {
        return;
    }

    ZnodeModelFormEditor editor = (ZnodeModelFormEditor) getEditor();
    Znode znode = znodeModel.getData();
    Stat stat = znode.getStat();

    if (!isDirty() || stat.getAversion() == editor.getLastModificationAversion()) {
        initFromModel();
    }
    else {
        editor.setActivePage(ID);
        setInfoText(EXTERNAL_MODIFICATION_INFO_TEXT);
    }

}

public void processResult(int rc, String path,
        Object ctx, byte[] data, Stat stat) {
    if (rc == KeeperException.Code.NONODE.intValue()) {
        // we can just ignore because the child watcher takes care of this
        return;
    }
    if (rc != KeeperException.Code.OK.intValue()) {
        zk.getData(myNode, (Watcher)ctx, this, ctx);
    }
    int currVer = stat.getVersion();
    if (currVer != lastVer) {
        String parts[] = new String(data).split(" ", 2);
        myInstance.configure(parts[1]);
        lastVer = currVer;
    }
}

private void recursiveCreate(ZooKeeper zk,String path,byte[] data) throws KeeperException, InterruptedException {
    if ("/".equals(path.trim())) return;
    path = StringUtils.trimTrailingCharacter(path, '/');
    Stack<String> paths = buildPathStack(path);
    byte[] tempdata = "".getBytes();
    while (!paths.empty()) {
        String elPath = paths.pop();
        Stat stat = zk.exists(elPath, false);
        if (paths.isEmpty()) tempdata = data;
        if (stat == null)zk.create(elPath, tempdata, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
}

@Override
public Stat exists(String path, boolean watch) throws KeeperException,
        InterruptedException {
    int count = 0;
    do {
        try {
            return super.exists(path, watch ? watcher : null);
        } catch (KeeperException.ConnectionLossException e) {
            LoggerFactory.getLogger().warn(
                    "ZooKeeper connection lost.  Trying to reconnect.");
        }
    } while (!closed && (limit == -1 || count++ < limit));
    return null;
}

static public void copyStat(Stat from, Stat to) {
    to.setAversion(from.getAversion());
    to.setCtime(from.getCtime());
    to.setCversion(from.getCversion());
    to.setCzxid(from.getCzxid());
    to.setMtime(from.getMtime());
    to.setMzxid(from.getMzxid());
    to.setVersion(from.getVersion());
    to.setEphemeralOwner(from.getEphemeralOwner());
    to.setDataLength(from.getDataLength());
    to.setNumChildren(from.getNumChildren());
}

/**
 * When authentication is enabled on Zookeeper, /hbase/root-region-server
 * should be created with 2 ACLs: one specifies that the hbase user has
 * full access to the node; the other, that it is world-readable.
 */
@Test (timeout=30000)
public void testHBaseRootRegionServerZNodeACL() throws Exception {
  if (!secureZKAvailable) {
    return;
  }

  List<ACL> acls = zkw.getRecoverableZooKeeper().getZooKeeper()
      .getACL("/hbase/root-region-server", new Stat());
  assertEquals(acls.size(),2);

  boolean foundWorldReadableAcl = false;
  boolean foundHBaseOwnerAcl = false;
  for(int i = 0; i < 2; i++) {
    if (acls.get(i).getId().getScheme().equals("world") == true) {
      assertEquals(acls.get(0).getId().getId(),"anyone");
      assertEquals(acls.get(0).getPerms(), ZooDefs.Perms.READ);
      foundWorldReadableAcl = true;
    }
    else {
      if (acls.get(i).getId().getScheme().equals("sasl") == true) {
        assertEquals(acls.get(1).getId().getId(),"hbase");
        assertEquals(acls.get(1).getId().getScheme(),"sasl");
        foundHBaseOwnerAcl = true;
      } else { // error: should not get here: test fails.
        assertTrue(false);
      }
    }
  }
  assertTrue(foundWorldReadableAcl);
  assertTrue(foundHBaseOwnerAcl);
}

/**
 * Updates data of an existing znode. The current content of the znode is passed to the {@link DataUpdater} that is
 * passed into this method, which returns the new content. The new content is only written back to ZooKeeper if
 * nobody has modified the given znode in between. If a concurrent change has been detected the new data of the
 * znode is passed to the updater once again until the new contents can be successfully written back to ZooKeeper.
 * 
 * @param <T>
 * @param path The path of the znode.
 * @param updater Updater that creates the new contents.
 */
public <T extends Object> void updateDataSerialized(String path, DataUpdater<T> updater) {
    Stat stat = new Stat();
    boolean retry;
    do {
        retry = false;
        try {
            T oldData = (T) readData(path, stat);
            T newData = updater.update(oldData);
            writeData(path, newData, stat.getVersion());
        } catch (ZkBadVersionException e) {
            retry = true;
        }
    } while (retry);
}

@GET
@Produces(MediaType.APPLICATION_OCTET_STREAM)
public Response getZNodeListAsOctet(@PathParam("path") String path)
        throws InterruptedException, KeeperException {
    ensurePathNotNull(path);

    Stat stat = new Stat();
    byte[] data = zk.getData(path, false, stat);

    if (data == null) {
        return Response.status(Response.Status.NO_CONTENT).build();
    } else {
        return Response.status(Response.Status.OK).entity(data).build();
    }
}

public int countChildren(String path) {
    try {
        Stat stat = new Stat();
        this.readData(path, stat);
        return stat.getNumChildren();
        //return getChildren(path).size();
    } catch (ZkNoNodeException e) {
        return -1;
    }
}

private byte[] getKey(ZooKeeperServer zks) {
    ZKDatabase db = zks.getZKDatabase();
    if (db != null) {
        try {
            Stat stat = new Stat();
            return db.getData("/key", stat, null);
        } catch (NoNodeException e) {
            LOG.error("getData failed", e);
        }
    }
    return null;
}

public byte[] getData(String path, Stat stat, Watcher watcher)
        throws KeeperException.NoNodeException {
    DataNode n = nodes.get(path);
    if (n == null) {
        throw new KeeperException.NoNodeException();
    }
    synchronized (n) {
        n.copyStat(stat);
        if (watcher != null) {
            dataWatches.addWatch(path, watcher);
        }
        return n.data;
    }
}

public static void waitForActiveLockData(TestContext ctx,
    ZooKeeperServer zks, String parentDir, byte[] activeData)
    throws Exception {
  long st = Time.now();
  long lastPrint = st;
  while (true) {
    if (ctx != null) {
      ctx.checkException();
    }
    try {
      Stat stat = new Stat();
      byte[] data = zks.getZKDatabase().getData(
        parentDir + "/" +
        ActiveStandbyElector.LOCK_FILENAME, stat, null);
      if (activeData != null &&
          Arrays.equals(activeData, data)) {
        return;
      }
      if (Time.now() > lastPrint + LOG_INTERVAL_MS) {
        LOG.info("Cur data: " + StringUtils.byteToHexString(data));
        lastPrint = Time.now();
      }
    } catch (NoNodeException nne) {
      if (activeData == null) {
        return;
      }
      if (Time.now() > lastPrint + LOG_INTERVAL_MS) {
        LOG.info("Cur data: no node");
        lastPrint = Time.now();
      }
    }
    Thread.sleep(50);
  }
}

public boolean isAllowsChildren(String nodePath) {
    if (connected) {
        try {
            Stat s = zooKeeper.exists(nodePath, false);
            if (s != null) {
                return s.getEphemeralOwner() == 0;
            }
        } catch (Exception e) {
            LoggerFactory.getLogger().error(
                    "Error occurred determining whether node is allowed children: "
                            + nodePath, e);
        }
    }
    return false;
}

@Test
public void testChild()
    throws IOException, KeeperException, InterruptedException
{
    String name = "/foo";
    zk.create(name, name.getBytes(), Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);

    String childname = name + "/bar";
    zk.create(childname, childname.getBytes(), Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL);

    Stat stat;

    stat = newStat();
    zk.getData(name, false, stat);

    Assert.assertEquals(stat.getCzxid(), stat.getMzxid());
    Assert.assertEquals(stat.getCzxid() + 1, stat.getPzxid());
    Assert.assertEquals(stat.getCtime(), stat.getMtime());
    Assert.assertEquals(1, stat.getCversion());
    Assert.assertEquals(0, stat.getVersion());
    Assert.assertEquals(0, stat.getAversion());
    Assert.assertEquals(0, stat.getEphemeralOwner());
    Assert.assertEquals(name.length(), stat.getDataLength());
    Assert.assertEquals(1, stat.getNumChildren());

    stat = newStat();
    zk.getData(childname, false, stat);

    Assert.assertEquals(stat.getCzxid(), stat.getMzxid());
    Assert.assertEquals(stat.getCzxid(), stat.getPzxid());
    Assert.assertEquals(stat.getCtime(), stat.getMtime());
    Assert.assertEquals(0, stat.getCversion());
    Assert.assertEquals(0, stat.getVersion());
    Assert.assertEquals(0, stat.getAversion());
    Assert.assertEquals(zk.getSessionId(), stat.getEphemeralOwner());
    Assert.assertEquals(childname.length(), stat.getDataLength());
    Assert.assertEquals(0, stat.getNumChildren());
}

@Override
public byte[] getData(String path, Watcher watcher, Stat stat)
        throws KeeperException, InterruptedException {
    int count = 0;
    do {
        try {
            return super.getData(path, watcher, stat);
        } catch (KeeperException.ConnectionLossException e) {
            LoggerFactory.getLogger().warn(
                    "ZooKeeper connection lost.  Trying to reconnect.");
        }
    } while (!closed && (limit == -1 || count++ < limit));
    return null;
}

/**
 * Expire the ZK session of the given service. This requires
 * (and asserts) that the given service be the current active.
 * @throws NoNodeException if no service holds the lock
 */
public void expireActiveLockHolder(int idx)
    throws NoNodeException {
  Stat stat = new Stat();
  byte[] data = zks.getZKDatabase().getData(
      DummyZKFC.LOCK_ZNODE, stat, null);

  assertArrayEquals(Ints.toByteArray(svcs.get(idx).index), data);
  long session = stat.getEphemeralOwner();
  LOG.info("Expiring svc " + idx + "'s zookeeper session " + session);
  zks.closeSession(session);
}

public void processResult(int rc, String path, Object ctx, byte[] data,
        Stat stat)
{
    this.data = data;
    this.stat = stat;
    super.processResult(Code.get(rc), path, ctx);
}