### Question:
ZookeeperDataSource extends AbstractDataSource<String, T> { @Override public void close() throws Exception { if (this.nodeCache != null) { this.nodeCache.getListenable().removeListener(listener); this.nodeCache.close(); } if (this.zkClient != null) { this.zkClient.close(); } pool.shutdown(); } ZookeeperDataSource(final String serverAddr, final String path, Converter<String, T> parser);  ZookeeperDataSource(final String serverAddr, final String groupId, final String dataId,
                               Converter<String, T> parser);  ZookeeperDataSource(final String serverAddr, final List<AuthInfo> authInfos, final String groupId, final String dataId,
                               Converter<String, T> parser); @Override String readSource(); @Override void close();  }### Answer:
@Test public void testZooKeeperDataSourceAuthorization() throws Exception { TestingServer server = new TestingServer(21812); server.start(); final String remoteAddress = server.getConnectString(); final String groupId = "sentinel-zk-ds-demo"; final String dataId = "flow-HK"; final String path = "/" + groupId + "/" + dataId; final String scheme = "digest"; final String auth = "root:123456"; AuthInfo authInfo = new AuthInfo(scheme, auth.getBytes()); List<AuthInfo> authInfoList = Collections.singletonList(authInfo); CuratorFramework zkClient = CuratorFrameworkFactory.builder(). connectString(remoteAddress). retryPolicy(new ExponentialBackoffRetry(3, 100)). authorization(authInfoList). build(); zkClient.start(); Stat stat = zkClient.checkExists().forPath(path); if (stat == null) { ACL acl = new ACL(ZooDefs.Perms.ALL, new Id(scheme, DigestAuthenticationProvider.generateDigest(auth))); zkClient.create().creatingParentContainersIfNeeded().withACL(Collections.singletonList(acl)).forPath(path, null); } ReadableDataSource<String, List<FlowRule>> flowRuleDataSource = new ZookeeperDataSource<List<FlowRule>>(remoteAddress, authInfoList, groupId, dataId, new Converter<String, List<FlowRule>>() { @Override public List<FlowRule> convert(String source) { return JSON.parseObject(source, new TypeReference<List<FlowRule>>() { }); } }); FlowRuleManager.register2Property(flowRuleDataSource.getProperty()); final String resourceName = "HK"; publishThenTestFor(zkClient, path, resourceName, 10); publishThenTestFor(zkClient, path, resourceName, 15); zkClient.close(); server.stop(); }