Randgalt commented on a change in pull request #3266:
URL: https://github.com/apache/hadoop/pull/3266#discussion_r683402550
##########
File path:
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java
##########
@@ -363,105 +357,122 @@ public void startThreads() throws IOException {
throw new RuntimeException("Could not create ZK paths");
}
try {
- keyCache = new PathChildrenCache(zkClient, ZK_DTSM_MASTER_KEY_ROOT,
true);
+ keyCache = CuratorCache.bridgeBuilder(zkClient, ZK_DTSM_MASTER_KEY_ROOT)
+ .build();
if (keyCache != null) {
- keyCache.start(StartMode.BUILD_INITIAL_CACHE);
- keyCache.getListenable().addListener(new PathChildrenCacheListener() {
- @Override
- public void childEvent(CuratorFramework client,
- PathChildrenCacheEvent event)
- throws Exception {
- switch (event.getType()) {
- case CHILD_ADDED:
- processKeyAddOrUpdate(event.getData().getData());
- break;
- case CHILD_UPDATED:
- processKeyAddOrUpdate(event.getData().getData());
- break;
- case CHILD_REMOVED:
- processKeyRemoved(event.getData().getPath());
- break;
- default:
- break;
- }
- }
- }, listenerThreadPool);
+ CuratorCacheListener keyCacheListener = CuratorCacheListener.builder()
+ .forCreates(childData -> {
+ try {
+ processKeyAddOrUpdate(childData.getData());
+ } catch (IOException e) {
+ LOG.error("Error while processing Curator keyCacheListener "
+ + "NODE_CREATED event");
+ throw new UncheckedIOException(e);
+ }
+ })
+ .forChanges((oldNode, node) -> {
+ try {
+ processKeyAddOrUpdate(node.getData());
+ } catch (IOException e) {
+ LOG.error("Error while processing Curator keyCacheListener "
+ + "NODE_CHANGED event");
+ throw new UncheckedIOException(e);
+ }
+ })
+ .forDeletes(childData -> processKeyRemoved(childData.getPath()))
+ .build();
+ keyCache.listenable().addListener(keyCacheListener);
+ keyCache.start();
loadFromZKCache(false);
}
} catch (Exception e) {
- throw new IOException("Could not start PathChildrenCache for keys", e);
+ throw new IOException("Could not start Curator keyCacheListener for
keys",
+ e);
}
if (isTokenWatcherEnabled) {
LOG.info("TokenCache is enabled");
try {
- tokenCache = new PathChildrenCache(zkClient, ZK_DTSM_TOKENS_ROOT,
true);
+ tokenCache = CuratorCache.bridgeBuilder(zkClient, ZK_DTSM_TOKENS_ROOT)
+ .build();
if (tokenCache != null) {
- tokenCache.start(StartMode.BUILD_INITIAL_CACHE);
- tokenCache.getListenable().addListener(new
PathChildrenCacheListener() {
-
- @Override
- public void childEvent(CuratorFramework client,
- PathChildrenCacheEvent event) throws
Exception {
- switch (event.getType()) {
- case CHILD_ADDED:
- processTokenAddOrUpdate(event.getData().getData());
- break;
- case CHILD_UPDATED:
- processTokenAddOrUpdate(event.getData().getData());
- break;
- case CHILD_REMOVED:
- processTokenRemoved(event.getData());
- break;
- default:
- break;
- }
- }
- }, listenerThreadPool);
+ CuratorCacheListener tokenCacheListener =
CuratorCacheListener.builder()
+ .forCreates(childData -> {
Review comment:
Note: there is a `forCreatesAndChanges()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-issues-h...@hadoop.apache.org
