@Test(timeout = 60000) public void testDumpEphemerals() throws Exception { int count = 1000; long session = 1000; long zxid = 2000; final DataTree dataTree = new DataTree(); LOG.info("Create {} zkclient sessions and its ephemeral nodes", count); createEphemeralNode(session, dataTree, count); final AtomicBoolean exceptionDuringDumpEphemerals = new AtomicBoolean( false); final AtomicBoolean running = new AtomicBoolean(true); Thread thread = new Thread() { public void run() { PrintWriter pwriter = new PrintWriter(new StringWriter()); try { while (running.get()) { dataTree.dumpEphemerals(pwriter); } } catch (Exception e) { LOG.error("Received exception while dumpEphemerals!", e); exceptionDuringDumpEphemerals.set(true); } }; }; thread.start(); LOG.debug("Killing {} zkclient sessions and its ephemeral nodes", count); killZkClientSession(session, zxid, dataTree, count); running.set(false); thread.join(); Assert.assertFalse("Should have got exception while dumpEphemerals!", exceptionDuringDumpEphemerals.get()); }
public void dumpEphemerals(PrintWriter pwriter) { pwriter.println("Sessions with Ephemerals (" + ephemerals.keySet().size() + "):"); for (Entry<Long, HashSet<String>> entry : ephemerals.entrySet()) { pwriter.print("0x" + Long.toHexString(entry.getKey())); pwriter.println(":"); HashSet<String> tmp = entry.getValue(); if (tmp != null) { synchronized (tmp) { for (String path : tmp) { pwriter.println("\t" + path); } } } } }
DataTree { public void dumpEphemerals(PrintWriter pwriter) { pwriter.println("Sessions with Ephemerals (" + ephemerals.keySet().size() + "):"); for (Entry<Long, HashSet<String>> entry : ephemerals.entrySet()) { pwriter.print("0x" + Long.toHexString(entry.getKey())); pwriter.println(":"); HashSet<String> tmp = entry.getValue(); if (tmp != null) { synchronized (tmp) { for (String path : tmp) { pwriter.println("\t" + path); } } } } } }
DataTree { public void dumpEphemerals(PrintWriter pwriter) { pwriter.println("Sessions with Ephemerals (" + ephemerals.keySet().size() + "):"); for (Entry<Long, HashSet<String>> entry : ephemerals.entrySet()) { pwriter.print("0x" + Long.toHexString(entry.getKey())); pwriter.println(":"); HashSet<String> tmp = entry.getValue(); if (tmp != null) { synchronized (tmp) { for (String path : tmp) { pwriter.println("\t" + path); } } } } } DataTree(); }
DataTree { public void dumpEphemerals(PrintWriter pwriter) { pwriter.println("Sessions with Ephemerals (" + ephemerals.keySet().size() + "):"); for (Entry<Long, HashSet<String>> entry : ephemerals.entrySet()) { pwriter.print("0x" + Long.toHexString(entry.getKey())); pwriter.println(":"); HashSet<String> tmp = entry.getValue(); if (tmp != null) { synchronized (tmp) { for (String path : tmp) { pwriter.println("\t" + path); } } } } } DataTree(); @SuppressWarnings("unchecked") Set<String> getEphemerals(long sessionId); Set<String> getContainers(); Set<String> getTtls(); Collection<Long> getSessions(); DataNode getNode(String path); int getNodeCount(); int getWatchCount(); int getEphemeralsCount(); long approximateDataSize(); void addConfigNode(); static void copyStatPersisted(StatPersisted from, StatPersisted to); static void copyStat(Stat from, Stat to); void updateCount(String lastPrefix, int diff); void updateBytes(String lastPrefix, long diff); void createNode(final String path, byte data[], List<ACL> acl,
            long ephemeralOwner, int parentCVersion, long zxid, long time); void createNode(final String path, byte data[], List<ACL> acl,
            long ephemeralOwner, int parentCVersion, long zxid, long time, Stat outputStat); void deleteNode(String path, long zxid); Stat setData(String path, byte data[], int version, long zxid,
            long time); String getMaxPrefixWithQuota(String path); byte[] getData(String path, Stat stat, Watcher watcher); Stat statNode(String path, Watcher watcher); List<String> getChildren(String path, Stat stat, Watcher watcher); Stat setACL(String path, List<ACL> acl, int version); List<ACL> getACL(String path, Stat stat); List<ACL> getACL(DataNode node); int aclCacheSize(); ProcessTxnResult processTxn(TxnHeader header, Record txn); ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn); void serializeNodeData(OutputArchive oa, String path, DataNode node); void serialize(OutputArchive oa, String tag); void deserialize(InputArchive ia, String tag); synchronized void dumpWatchesSummary(PrintWriter pwriter); synchronized void dumpWatches(PrintWriter pwriter, boolean byPath); synchronized WatchesReport getWatches(); synchronized WatchesPathReport getWatchesByPath(); synchronized WatchesSummary getWatchesSummary(); void dumpEphemerals(PrintWriter pwriter); Map<Long, Set<String>> getEphemerals(); void removeCnxn(Watcher watcher); void setWatches(long relativeZxid, List<String> dataWatches,
            List<String> existWatches, List<String> childWatches,
            Watcher watcher); void setCversionPzxid(String path, int newCversion, long zxid); boolean containsWatcher(String path, WatcherType type, Watcher watcher); boolean removeWatch(String path, WatcherType type, Watcher watcher); ReferenceCountedACLCache getReferenceCountedAclCache(); }
DataTree { public void dumpEphemerals(PrintWriter pwriter) { pwriter.println("Sessions with Ephemerals (" + ephemerals.keySet().size() + "):"); for (Entry<Long, HashSet<String>> entry : ephemerals.entrySet()) { pwriter.print("0x" + Long.toHexString(entry.getKey())); pwriter.println(":"); HashSet<String> tmp = entry.getValue(); if (tmp != null) { synchronized (tmp) { for (String path : tmp) { pwriter.println("\t" + path); } } } } } DataTree(); @SuppressWarnings("unchecked") Set<String> getEphemerals(long sessionId); Set<String> getContainers(); Set<String> getTtls(); Collection<Long> getSessions(); DataNode getNode(String path); int getNodeCount(); int getWatchCount(); int getEphemeralsCount(); long approximateDataSize(); void addConfigNode(); static void copyStatPersisted(StatPersisted from, StatPersisted to); static void copyStat(Stat from, Stat to); void updateCount(String lastPrefix, int diff); void updateBytes(String lastPrefix, long diff); void createNode(final String path, byte data[], List<ACL> acl,
            long ephemeralOwner, int parentCVersion, long zxid, long time); void createNode(final String path, byte data[], List<ACL> acl,
            long ephemeralOwner, int parentCVersion, long zxid, long time, Stat outputStat); void deleteNode(String path, long zxid); Stat setData(String path, byte data[], int version, long zxid,
            long time); String getMaxPrefixWithQuota(String path); byte[] getData(String path, Stat stat, Watcher watcher); Stat statNode(String path, Watcher watcher); List<String> getChildren(String path, Stat stat, Watcher watcher); Stat setACL(String path, List<ACL> acl, int version); List<ACL> getACL(String path, Stat stat); List<ACL> getACL(DataNode node); int aclCacheSize(); ProcessTxnResult processTxn(TxnHeader header, Record txn); ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn); void serializeNodeData(OutputArchive oa, String path, DataNode node); void serialize(OutputArchive oa, String tag); void deserialize(InputArchive ia, String tag); synchronized void dumpWatchesSummary(PrintWriter pwriter); synchronized void dumpWatches(PrintWriter pwriter, boolean byPath); synchronized WatchesReport getWatches(); synchronized WatchesPathReport getWatchesByPath(); synchronized WatchesSummary getWatchesSummary(); void dumpEphemerals(PrintWriter pwriter); Map<Long, Set<String>> getEphemerals(); void removeCnxn(Watcher watcher); void setWatches(long relativeZxid, List<String> dataWatches,
            List<String> existWatches, List<String> childWatches,
            Watcher watcher); void setCversionPzxid(String path, int newCversion, long zxid); boolean containsWatcher(String path, WatcherType type, Watcher watcher); boolean removeWatch(String path, WatcherType type, Watcher watcher); ReferenceCountedACLCache getReferenceCountedAclCache(); volatile long lastProcessedZxid; }