### Question:
ZKLogStreamMetadataStore implements LogStreamMetadataStore { @Override public void close() throws IOException { this.zooKeeperClient.unregister(permitManager); this.permitManager.close(); this.logSegmentStore.close(); SchedulerUtils.shutdownScheduler( getLockStateExecutor(false), conf.getSchedulerShutdownTimeoutMs(), TimeUnit.MILLISECONDS); } 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:
@SuppressWarnings("deprecation") @Test(timeout = 60000) public void testCreateLogMetadataWithCustomMetadata() throws Exception { String logName = testName.getMethodName(); String logIdentifier = "<default>"; List<String> pathsToDelete = Lists.newArrayList(); DLMetadata.create(new BKDLConfig(zkServers, "/ledgers")).update(uri); Namespace namespace = NamespaceBuilder.newBuilder() .conf(new DistributedLogConfiguration()) .uri(uri) .build(); org.apache.distributedlog.api.MetadataAccessor accessor = namespace.getNamespaceDriver().getMetadataAccessor(logName); accessor.createOrUpdateMetadata(logName.getBytes("UTF-8")); accessor.close(); testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, false); }
@Test(timeout = 60000) public void testCreateLogMetadataWithCustomMetadata() throws Exception { String logName = testName.getMethodName(); String logIdentifier = "<default>"; List<String> pathsToDelete = Lists.newArrayList(); DLMetadata.create(new BKDLConfig(zkServers, "/ledgers")).update(uri); Namespace namespace = NamespaceBuilder.newBuilder() .conf(new DistributedLogConfiguration()) .uri(uri) .build(); MetadataAccessor accessor = namespace.getNamespaceDriver().getMetadataAccessor(logName); accessor.createOrUpdateMetadata(logName.getBytes("UTF-8")); accessor.close(); testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, false); }