### Question:
ZKLogStreamMetadataStore implements LogStreamMetadataStore { static CompletableFuture<List<Versioned<byte[]>>> checkLogMetadataPaths(ZooKeeper zk, String logRootPath, boolean ownAllocator) { final String logRootParentPath = Utils.getParent(logRootPath); final String logSegmentsPath = logRootPath + LOGSEGMENTS_PATH; final String maxTxIdPath = logRootPath + MAX_TXID_PATH; final String lockPath = logRootPath + LOCK_PATH; final String readLockPath = logRootPath + READ_LOCK_PATH; final String versionPath = logRootPath + VERSION_PATH; final String allocationPath = logRootPath + ALLOCATION_PATH; int numPaths = ownAllocator ? MetadataIndex.ALLOCATION + 1 : MetadataIndex.LOGSEGMENTS + 1; List<CompletableFuture<Versioned<byte[]>>> checkFutures = Lists.newArrayListWithExpectedSize(numPaths); checkFutures.add(Utils.zkGetData(zk, logRootParentPath, false)); checkFutures.add(Utils.zkGetData(zk, logRootPath, false)); checkFutures.add(Utils.zkGetData(zk, maxTxIdPath, false)); checkFutures.add(Utils.zkGetData(zk, versionPath, false)); checkFutures.add(Utils.zkGetData(zk, lockPath, false)); checkFutures.add(Utils.zkGetData(zk, readLockPath, false)); checkFutures.add(Utils.zkGetData(zk, logSegmentsPath, false)); if (ownAllocator) { checkFutures.add(Utils.zkGetData(zk, allocationPath, false)); } return FutureUtils.collect(checkFutures); } ZKLogStreamMetadataStore(String clientId,
                                    DistributedLogConfiguration conf,
                                    ZooKeeperClient zkc,
                                    OrderedScheduler scheduler,
                                    StatsLogger statsLogger); @Override void close(); @Override LogSegmentMetadataStore getLogSegmentMetadataStore(); @Override PermitManager getPermitManager(); @Override Transaction<Object> newTransaction(); @Override CompletableFuture<Void> logExists(URI uri, final String logName); @Override DistributedLock createWriteLock(LogMetadataForWriter metadata); @Override CompletableFuture<DistributedLock> createReadLock(final LogMetadataForReader metadata,
                                                  Optional<String> readerId); @Override CompletableFuture<LogMetadataForWriter> getLog(final URI uri,
                                               final String logName,
                                               final boolean ownAllocator,
                                               final boolean createIfNotExists); @Override CompletableFuture<Void> deleteLog(URI uri, final String logName); @Override CompletableFuture<Void> renameLog(URI uri, String oldStreamName, String newStreamName);  }### Answer:
@Test(timeout = 60000) public void testCheckLogMetadataPathsWithAllocator() throws Exception { String logRootPath = "/" + testName.getMethodName(); List<Versioned<byte[]>> metadatas = Utils.ioResult(checkLogMetadataPaths( zkc.get(), logRootPath, true)); assertEquals("Should have 8 paths", 8, metadatas.size()); for (Versioned<byte[]> path : metadatas.subList(2, metadatas.size())) { assertNull(path.getValue()); assertNull(path.getVersion()); } }
@Test(timeout = 60000) public void testCheckLogMetadataPathsWithoutAllocator() throws Exception { String logRootPath = "/" + testName.getMethodName(); List<Versioned<byte[]>> metadatas = Utils.ioResult(checkLogMetadataPaths( zkc.get(), logRootPath, false)); assertEquals("Should have 7 paths", 7, metadatas.size()); for (Versioned<byte[]> path : metadatas.subList(2, metadatas.size())) { assertNull(path.getValue()); assertNull(path.getVersion()); } }