BKDistributedLogNamespace implements DistributedLogNamespace { public static Builder newBuilder() { return new Builder(); } 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 testAclPermsZkAccessNoConflict() throws Exception { String namespace = "/" + runtime.getMethodName(); initDlogMeta(namespace, "test-un", "test-stream"); URI uri = createDLMURI(namespace); ZooKeeperClient zkc = TestZooKeeperClientBuilder.newBuilder() .name("unpriv") .uri(uri) .build(); zkc.get().getChildren(uri.getPath() + "/test-stream", false, new Stat()); zkc.get().getData(uri.getPath() + "/test-stream", false, new Stat()); }
@Test(timeout = 60000) public void testAclPermsZkAccessConflict() throws Exception { String namespace = "/" + runtime.getMethodName(); initDlogMeta(namespace, "test-un", "test-stream"); URI uri = createDLMURI(namespace); ZooKeeperClient zkc = TestZooKeeperClientBuilder.newBuilder() .name("unpriv") .uri(uri) .build(); try { zkc.get().create(uri.getPath() + "/test-stream/test-garbage", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); fail("write should have failed due to perms"); } catch (KeeperException.NoAuthException ex) { LOG.info("caught exception trying to write with no perms", ex); } try { zkc.get().setData(uri.getPath() + "/test-stream", new byte[0], 0); fail("write should have failed due to perms"); } catch (KeeperException.NoAuthException ex) { LOG.info("caught exception trying to write with no perms", ex); } }