ZKAccessControlManager implements AccessControlManager, Watcher { public ZKAccessControlManager(DistributedLogConfiguration conf, ZooKeeperClient zkc, String zkRootPath, ScheduledExecutorService scheduledExecutorService) throws IOException { this.conf = conf; this.zkc = zkc; this.zkRootPath = zkRootPath; this.scheduledExecutorService = scheduledExecutorService; this.streamEntries = new ConcurrentHashMap<String, ZKAccessControl>(); try { Await.result(fetchDefaultAccessControlEntry()); } catch (Throwable t) { if (t instanceof InterruptedException) { throw new DLInterruptedException("Interrupted on getting default access control entry for " + zkRootPath, t); } else if (t instanceof KeeperException) { throw new IOException("Encountered zookeeper exception on getting default access control entry for " + zkRootPath, t); } else if (t instanceof IOException) { throw (IOException) t; } else { throw new IOException("Encountered unknown exception on getting access control entries for " + zkRootPath, t); } } try { Await.result(fetchAccessControlEntries()); } catch (Throwable t) { if (t instanceof InterruptedException) { throw new DLInterruptedException("Interrupted on getting access control entries for " + zkRootPath, t); } else if (t instanceof KeeperException) { throw new IOException("Encountered zookeeper exception on getting access control entries for " + zkRootPath, t); } else if (t instanceof IOException) { throw (IOException) t; } else { throw new IOException("Encountered unknown exception on getting access control entries for " + zkRootPath, t); } } } ZKAccessControlManager(DistributedLogConfiguration conf,
                                  ZooKeeperClient zkc,
                                  String zkRootPath,
                                  ScheduledExecutorService scheduledExecutorService); @Override boolean allowWrite(String stream); @Override boolean allowTruncate(String stream); @Override boolean allowDelete(String stream); @Override boolean allowAcquire(String stream); @Override boolean allowRelease(String stream); @Override void close(); @Override void process(WatchedEvent event);  }
@Test(timeout = 60000) public void testZKAccessControlManager() throws Exception { String zkRootPath = "/test-zk-access-control-manager"; String stream1 = "test-acm-1"; String stream2 = "test-acm-2"; logger.info("Creating ACL Manager for {}", zkRootPath); ZKAccessControlManager zkcm = new ZKAccessControlManager(conf, zkc, zkRootPath, executorService); logger.info("Created ACL Manager for {}", zkRootPath); try { verifyStreamPermissions(zkcm, stream1, true, true, true, true, true); String zkPath1 = zkRootPath + "/" + stream1; AccessControlEntry ace1 = new AccessControlEntry(); ace1.setDenyDelete(true); ZKAccessControl accessControl1 = new ZKAccessControl(ace1, zkPath1); setACL(accessControl1); logger.info("Create ACL for stream {} : {}", stream1, accessControl1); while (zkcm.allowDelete(stream1)) { Thread.sleep(100); } verifyStreamPermissions(zkcm, stream1, true, true, true, false, true); ace1 = new AccessControlEntry(); ace1.setDenyWrite(true); accessControl1 = new ZKAccessControl(ace1, zkPath1); setACL(accessControl1); logger.info("Update ACL for stream {} : {}", stream1, accessControl1); String zkPath2 = zkRootPath + "/" + stream2; AccessControlEntry ace2 = new AccessControlEntry(); ace2.setDenyTruncate(true); ZKAccessControl accessControl2 = new ZKAccessControl(ace2, zkPath2); setACL(accessControl2); logger.info("Create ACL for stream {} : {}", stream2, accessControl2); while (zkcm.allowWrite(stream1)) { Thread.sleep(100); } while (zkcm.allowTruncate(stream2)) { Thread.sleep(100); } verifyStreamPermissions(zkcm, stream1, false, true, true, true, true); verifyStreamPermissions(zkcm, stream2, true, false, true, true, true); Await.result(ZKAccessControl.delete(zkc, zkPath2)); logger.info("Delete ACL for stream {}", stream2); while (!zkcm.allowTruncate(stream2)) { Thread.sleep(100); } verifyStreamPermissions(zkcm, stream1, false, true, true, true, true); verifyStreamPermissions(zkcm, stream2, true, true, true, true, true); ZooKeeperClientUtils.expireSession(zkc, zkServers, 1000); ace1 = new AccessControlEntry(); ace1.setDenyRelease(true); accessControl1 = new ZKAccessControl(ace1, zkPath1); setACL(accessControl1); logger.info("Update ACL for stream {} : {}", stream1, accessControl1); ace2 = new AccessControlEntry(); ace2.setDenyAcquire(true); accessControl2 = new ZKAccessControl(ace2, zkPath2); setACL(accessControl2); logger.info("Created ACL for stream {} again : {}", stream2, accessControl2); while (zkcm.allowRelease(stream1)) { Thread.sleep(100); } while (zkcm.allowAcquire(stream2)) { Thread.sleep(100); } verifyStreamPermissions(zkcm, stream1, true, true, false, true, true); verifyStreamPermissions(zkcm, stream2, true, true, true, true, false); } finally { zkcm.close(); } }