FederatedZKLogMetadataStore extends NamespaceWatcher implements LogMetadataStore, Watcher, Runnable,
        FutureEventListener<Set<URI>> { @Override public Future<Iterator<String>> getLogs() { if (duplicatedLogFound.get()) { return duplicatedLogException(duplicatedLogName.get()); } return postStateCheck(retrieveLogs().map( new AbstractFunction1<List<Set<String>>, Iterator<String>>() { @Override public Iterator<String> apply(List<Set<String>> resultList) { return getIterator(resultList); } })); } FederatedZKLogMetadataStore(
            DistributedLogConfiguration conf,
            URI namespace,
            ZooKeeperClient zkc,
            OrderedScheduler scheduler); static void createFederatedNamespace(URI namespace, ZooKeeperClient zkc); @Override void run(); @Override void onSuccess(Set<URI> uris); @Override void onFailure(Throwable cause); @Override void process(WatchedEvent watchedEvent); @Override Future<URI> createLog(final String logName); @Override Future<Optional<URI>> getLogLocation(final String logName); @Override Future<Iterator<String>> getLogs(); @Override void registerNamespaceListener(NamespaceListener listener);  }
@Test(timeout = 60000) public void testGetLogs() throws Exception { int numLogs = 3 * maxLogsPerSubnamespace; Set<String> expectedLogs = createLogs(numLogs, "test-get-logs"); Set<String> receivedLogs; do { TimeUnit.MILLISECONDS.sleep(20); receivedLogs = new TreeSet<String>(); Iterator<String> logs = FutureUtils.result(metadataStore.getLogs()); receivedLogs.addAll(Lists.newArrayList(logs)); } while (receivedLogs.size() < numLogs); assertEquals(numLogs, receivedLogs.size()); assertTrue(Sets.difference(expectedLogs, receivedLogs).isEmpty()); }