### Question:
ZKLogMetadataStore implements LogMetadataStore { @Override public CompletableFuture<Iterator<String>> getLogs(String logNamePrefix) { final CompletableFuture<Iterator<String>> promise = new CompletableFuture<Iterator<String>>(); final String nsRootPath; if (StringUtils.isEmpty(logNamePrefix)) { nsRootPath = namespace.getPath(); } else { nsRootPath = namespace.getPath() + "/" + logNamePrefix; } try { final ZooKeeper zk = zkc.get(); zk.sync(nsRootPath, new AsyncCallback.VoidCallback() { @Override public void processResult(int syncRc, String syncPath, Object ctx) { if (KeeperException.Code.OK.intValue() == syncRc) { zk.getChildren(nsRootPath, false, new AsyncCallback.Children2Callback() { @Override public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) { if (KeeperException.Code.OK.intValue() == rc) { List<String> results = Lists.newArrayListWithExpectedSize(children.size()); for (String child : children) { if (!isReservedStreamName(child)) { results.add(child); } } promise.complete(results.iterator()); } else if (KeeperException.Code.NONODE.intValue() == rc) { List<String> streams = Lists.newLinkedList(); promise.complete(streams.iterator()); } else { promise.completeExceptionally(new ZKException("Error reading namespace " + nsRootPath, KeeperException.Code.get(rc))); } } }, null); } else if (KeeperException.Code.NONODE.intValue() == syncRc) { List<String> streams = Lists.newLinkedList(); promise.complete(streams.iterator()); } else { promise.completeExceptionally(new ZKException("Error reading namespace " + nsRootPath, KeeperException.Code.get(syncRc))); } } }, null); zkc.get(); } catch (ZooKeeperClient.ZooKeeperConnectionException e) { promise.completeExceptionally(e); } catch (InterruptedException e) { Thread.currentThread().interrupt(); promise.completeExceptionally(e); } return promise; } ZKLogMetadataStore(
            DistributedLogConfiguration conf,
            URI namespace,
            ZooKeeperClient zkc,
            OrderedScheduler scheduler); @Override CompletableFuture<URI> createLog(String logName); @Override CompletableFuture<Optional<URI>> getLogLocation(String logName); @Override CompletableFuture<Iterator<String>> getLogs(String logNamePrefix); @Override void registerNamespaceListener(NamespaceListener listener);  }### Answer:
@Test(timeout = 60000) public void testGetLogs() throws Exception { Set<String> logs = Sets.newHashSet(); for (int i = 0; i < 10; i++) { String logName = "test-" + i; logs.add(logName); createLogInNamespace(uri, logName); } Set<String> result = Sets.newHashSet(Utils.ioResult(metadataStore.getLogs(""))); assertEquals(10, result.size()); assertTrue(Sets.difference(logs, result).isEmpty()); }
@Test(timeout = 60000) public void testGetLogsPrefix() throws Exception { Set<String> logs = Sets.newHashSet(); for (int i = 0; i < 10; i++) { String logName = "test-" + i; logs.add(logName); createLogInNamespace(uri, "test/" + logName); } Set<String> result = Sets.newHashSet(Utils.ioResult(metadataStore.getLogs("test"))); assertEquals(10, result.size()); assertTrue(Sets.difference(logs, result).isEmpty()); }