### Question:
ZKMetadataDriverBase implements AutoCloseable { @SuppressWarnings("deprecation") @SneakyThrows(InterruptedException.class) protected void initialize(AbstractConfiguration<?> conf, StatsLogger statsLogger, RetryPolicy zkRetryPolicy, Optional<Object> optionalCtx) throws MetadataException { this.conf = conf; this.acls = ZkUtils.getACLs(conf); if (optionalCtx.isPresent() && optionalCtx.get() instanceof ZooKeeper) { this.ledgersRootPath = conf.getZkLedgersRootPath(); log.info("Initialize zookeeper metadata driver with external zookeeper client : ledgersRootPath = {}.", ledgersRootPath); this.zk = (ZooKeeper) (optionalCtx.get()); this.ownZKHandle = false; } else { final String metadataServiceUriStr; try { metadataServiceUriStr = conf.getMetadataServiceUri(); } catch (ConfigurationException e) { log.error("Failed to retrieve metadata service uri from configuration", e); throw new MetadataException( Code.INVALID_METADATA_SERVICE_URI, e); } URI metadataServiceUri = URI.create(metadataServiceUriStr); this.ledgersRootPath = metadataServiceUri.getPath(); final String bookieRegistrationPath = ledgersRootPath + "/" + AVAILABLE_NODE; final String bookieReadonlyRegistrationPath = bookieRegistrationPath + "/" + READONLY; final String zkServers = getZKServersFromServiceUri(metadataServiceUri); log.info("Initialize zookeeper metadata driver at metadata service uri {} :" + " zkServers = {}, ledgersRootPath = {}.", metadataServiceUriStr, zkServers, ledgersRootPath); try { this.zk = ZooKeeperClient.newBuilder() .connectString(zkServers) .sessionTimeoutMs(conf.getZkTimeout()) .operationRetryPolicy(zkRetryPolicy) .requestRateLimit(conf.getZkRequestRateLimit()) .statsLogger(statsLogger) .build(); if (null == zk.exists(bookieReadonlyRegistrationPath, false)) { try { zk.create(bookieReadonlyRegistrationPath, EMPTY_BYTE_ARRAY, acls, CreateMode.PERSISTENT); } catch (KeeperException.NodeExistsException e) { } catch (KeeperException.NoNodeException e) { } } } catch (IOException | KeeperException e) { log.error("Failed to create zookeeper client to {}", zkServers, e); MetadataException me = new MetadataException( Code.METADATA_SERVICE_ERROR, "Failed to create zookeeper client to " + zkServers, e); me.fillInStackTrace(); throw me; } this.ownZKHandle = true; } this.layoutManager = new ZkLayoutManager( zk, ledgersRootPath, acls); }  static String getZKServersFromServiceUri(URI uri); @SuppressWarnings("deprecation") static String resolveZkServers(AbstractConfiguration<?> conf); @SuppressWarnings("deprecation") static String resolveZkLedgersRootPath(AbstractConfiguration<?> conf); @SuppressWarnings("deprecation") static Class<? extends LedgerManagerFactory> resolveLedgerManagerFactory(URI metadataServiceUri); String getScheme(); LayoutManager getLayoutManager(); @SneakyThrows synchronized LedgerManagerFactory getLedgerManagerFactory(); @Override void close();  }### Answer:
@Test public void testInitialize() throws Exception { driver.initialize( conf, NullStatsLogger.INSTANCE, retryPolicy, Optional.empty()); assertEquals( "/path/to/ledgers", driver.ledgersRootPath); assertTrue(driver.ownZKHandle); String readonlyPath = "/path/to/ledgers/" + AVAILABLE_NODE + "/" + READONLY; assertSame(mockZkc, driver.zk); verifyStatic(ZooKeeperClient.class, times(1)); ZooKeeperClient.newBuilder(); verify(mockZkBuilder, times(1)).build(); verify(mockZkc, times(1)) .exists(eq(readonlyPath), eq(false)); assertNotNull(driver.layoutManager); assertNull(driver.lmFactory); driver.close(); verify(mockZkc, times(1)).close(); assertNull(driver.zk); }