BKDistributedLogNamespace implements DistributedLogNamespace { static String validateAndGetFullLedgerAllocatorPoolPath(DistributedLogConfiguration conf, URI uri) throws IOException { String poolPath = conf.getLedgerAllocatorPoolPath(); LOG.info("PoolPath is {}", poolPath); if (null == poolPath || !poolPath.startsWith(".") || poolPath.endsWith("/")) { LOG.error("Invalid ledger allocator pool path specified when enabling ledger allocator pool : {}", poolPath); throw new IOException("Invalid ledger allocator pool path specified : " + poolPath); } String poolName = conf.getLedgerAllocatorPoolName(); if (null == poolName) { LOG.error("No ledger allocator pool name specified when enabling ledger allocator pool."); throw new IOException("No ledger allocator name specified when enabling ledger allocator pool."); } String rootPath = uri.getPath() + "/" + poolPath + "/" + poolName; try { PathUtils.validatePath(rootPath); } catch (IllegalArgumentException iae) { LOG.error("Invalid ledger allocator pool path specified when enabling ledger allocator pool : {}", poolPath); throw new IOException("Invalid ledger allocator pool path specified : " + poolPath); } return rootPath; } private  BKDistributedLogNamespace(
            DistributedLogConfiguration conf,
            URI uri,
            FeatureProvider featureProvider,
            StatsLogger statsLogger,
            StatsLogger perLogStatsLogger,
            String clientId,
            int regionId,
            ZooKeeperClientBuilder nsZkcBuilder,
            ZooKeeperClient nsZkc,
            BKDLConfig bkdlConfig); static Builder newBuilder(); @Override void createLog(String logName); @Override void deleteLog(String logName); @Override DistributedLogManager openLog(String logName); @Override DistributedLogManager openLog(String logName,
                                         Optional<DistributedLogConfiguration> logConf,
                                         Optional<DynamicDistributedLogConfiguration> dynamicLogConf); @Override boolean logExists(String logName); @Override Iterator<String> getLogs(); @Override void registerNamespaceListener(NamespaceListener listener); @Override synchronized AccessControlManager createAccessControlManager(); @VisibleForTesting ZooKeeperClient getSharedWriterZKCForDL(); @VisibleForTesting BookKeeperClient getReaderBKC(); @VisibleForTesting LogSegmentMetadataStore getWriterSegmentMetadataStore(); @VisibleForTesting LedgerAllocator getLedgerAllocator(); DistributedLogManager createDistributedLogManagerWithSharedClients(String nameOfLogStream); DistributedLogManager createDistributedLogManager(
            String nameOfLogStream,
            ClientSharingOption clientSharingOption); DistributedLogManager createDistributedLogManager(
            String nameOfLogStream,
            ClientSharingOption clientSharingOption,
            Optional<DistributedLogConfiguration> logConfiguration,
            Optional<DynamicDistributedLogConfiguration> dynamicLogConfiguration); MetadataAccessor createMetadataAccessor(String nameOfMetadataNode); Collection<String> enumerateAllLogsInNamespace(); Map<String, byte[]> enumerateLogsWithMetadataInNamespace(); static Map<String, byte[]> enumerateLogsWithMetadataInNamespace(final DistributedLogConfiguration conf, final URI uri); @Override void close();  }
@Test(timeout = 60000) public void testValidateAndGetFullLedgerAllocatorPoolPath() throws Exception { DistributedLogConfiguration testConf = new DistributedLogConfiguration(); testConf.setEnableLedgerAllocatorPool(true); String namespace = "/" + runtime.getMethodName(); URI uri = createDLMURI(namespace); testConf.setLedgerAllocatorPoolName("test"); testConf.setLedgerAllocatorPoolPath("test"); validateBadAllocatorConfiguration(testConf, uri); testConf.setLedgerAllocatorPoolPath("."); validateBadAllocatorConfiguration(testConf, uri); testConf.setLedgerAllocatorPoolPath(".."); validateBadAllocatorConfiguration(testConf, uri); testConf.setLedgerAllocatorPoolPath("./"); validateBadAllocatorConfiguration(testConf, uri); testConf.setLedgerAllocatorPoolPath(".test/"); validateBadAllocatorConfiguration(testConf, uri); testConf.setLedgerAllocatorPoolPath(".test"); testConf.setLedgerAllocatorPoolName(null); validateBadAllocatorConfiguration(testConf, uri); }