From 4aa7bdc18f137c1e8844d065ebf161f31744a1c1 Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Mon, 17 Aug 2026 21:45:27 +0800 Subject: [PATCH 1/3] HADOOP-19966. ZKDelegationTokenSecretManager may fail to load tokens and keys from ZooKeeper on startup Wait for CuratorCache initialization before loading each cache, skip the root container znode when processing cache events, and create the container znodes with explicit empty data. Assisted-by: deepseek-v4-pro --- .../ZKDelegationTokenSecretManager.java | 45 +++++++- .../TestZKDelegationTokenSecretManager.java | 100 ++++++++++-------- 2 files changed, 102 insertions(+), 43 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java index dd2ab3ff26f76d..65068bc76128d2 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java @@ -24,6 +24,8 @@ import java.io.DataOutputStream; import java.io.IOException; import java.io.UncheckedIOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Stream; @@ -119,6 +121,8 @@ public abstract class ZKDelegationTokenSecretManager { + if (ZK_DTSM_MASTER_KEY_ROOT.equals(node.getPath())) { + // The root node itself is a container, not a key. + return; + } try { processKeyAddOrUpdate(node.getData()); } catch (IOException e) { @@ -301,9 +310,11 @@ public void startThreads() throws IOException { } }) .forDeletes(childData -> processKeyRemoved(childData.getPath())) + .forInitialized(keyCacheInitialized::countDown) .build(); keyCache.listenable().addListener(keyCacheListener); keyCache.start(); + awaitCacheInitialized(keyCacheInitialized, "key"); loadFromZKCache(false); } catch (Exception e) { throw new IOException("Could not start Curator keyCacheListener for keys", @@ -314,8 +325,13 @@ public void startThreads() throws IOException { try { tokenCache = CuratorCache.bridgeBuilder(zkClient, ZK_DTSM_TOKENS_ROOT) .build(); + CountDownLatch tokenCacheInitialized = new CountDownLatch(1); CuratorCacheListener tokenCacheListener = CuratorCacheListener.builder() .forCreatesAndChanges((oldNode, node) -> { + if (ZK_DTSM_TOKENS_ROOT.equals(node.getPath())) { + // The root node itself is a container, not a token. + return; + } try { processTokenAddOrUpdate(node.getData()); } catch (IOException e) { @@ -333,9 +349,11 @@ public void startThreads() throws IOException { throw new UncheckedIOException(e); } }) + .forInitialized(tokenCacheInitialized::countDown) .build(); tokenCache.listenable().addListener(tokenCacheListener); tokenCache.start(); + awaitCacheInitialized(tokenCacheInitialized, "token"); loadFromZKCache(true); } catch (Exception e) { throw new IOException( @@ -363,6 +381,11 @@ private void loadFromZKCache(final boolean isTokenCache) { final AtomicInteger count = new AtomicInteger(0); children.forEach(childData -> { + if (childData.getPath().equals( + isTokenCache ? ZK_DTSM_TOKENS_ROOT : ZK_DTSM_MASTER_KEY_ROOT)) { + // The root node itself is a container, not a key or token. + return; + } try { if (isTokenCache) { processTokenAddOrUpdate(childData.getData()); @@ -386,6 +409,20 @@ private void loadFromZKCache(final boolean isTokenCache) { LOG.info("Loaded {} cache.", cacheName); } + private void awaitCacheInitialized(CountDownLatch initialized, + String cacheName) throws IOException { + try { + if (!initialized.await(CACHE_INITIALIZED_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + throw new IOException("Timed out waiting for " + cacheName + + " cache initialization"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while waiting for " + cacheName + + " cache initialization", e); + } + } + private void processKeyAddOrUpdate(byte[] data) throws IOException { ByteArrayInputStream bin = new ByteArrayInputStream(data); DataInputStream din = new DataInputStream(bin); @@ -425,6 +462,10 @@ protected TokenIdent processTokenAddOrUpdate(byte[] data) throws IOException { } private void processTokenRemoved(ChildData data) throws IOException { + if (ZK_DTSM_TOKENS_ROOT.equals(data.getPath())) { + // The root node itself is a container, not a token. + return; + } ByteArrayInputStream bin = new ByteArrayInputStream(data.getData()); DataInputStream din = new DataInputStream(bin); TokenIdent ident = createIdentifier(); @@ -474,7 +515,9 @@ public void stopThreads() { private void createPersistentNode(String nodePath) throws Exception { try { - zkClient.create().withMode(CreateMode.PERSISTENT).forPath(nodePath); + // Pass empty data explicitly; Curator 5.2.0+ defaults data-less creates + // to the local address, which breaks parsing of the container nodes. + zkClient.create().withMode(CreateMode.PERSISTENT).forPath(nodePath, new byte[0]); } catch (KeeperException.NodeExistsException ne) { LOG.debug(nodePath + " znode already exists !!"); } catch (Exception e) { diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java index 58c3f541018411..9d70e54d6d24ec 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java @@ -522,6 +522,13 @@ public Boolean get() { // The good token should be loaded on startup, and removed after expiry. id = smNew.decodeTokenIdentifier(token); + final AbstractDelegationTokenIdentifier idGood = id; + GenericTestUtils.waitFor(new Supplier() { + @Override + public Boolean get() { + return zksmNew.getTokenInfoFromMemory(idGood) != null; + } + }, 100, 5000); dtinfo = zksmNew.getTokenInfoFromMemory(id); assertNotNull(dtinfo, "good dt should be in memory!"); @@ -551,22 +558,27 @@ public void testCreatingParentContainersIfNeeded() throws Exception { .build(); curatorFramework.start(); ZKDelegationTokenSecretManager.setCurator(curatorFramework); - DelegationTokenManager tm1 = new DelegationTokenManager(conf, new Text("foo")); + try { + DelegationTokenManager tm1 = new DelegationTokenManager(conf, new Text("foo")); - // When the init method is called, - // the ZKDelegationTokenSecretManager#startThread method will be called, - // and the creatingParentContainersIfNeeded will be called to create the nameSpace. - tm1.init(); + // When the init method is called, + // the ZKDelegationTokenSecretManager#startThread method will be called, + // and the creatingParentContainersIfNeeded will be called to create the nameSpace. + tm1.init(); - String workingPath = "/" + conf.get(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH, - ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH_DEAFULT) + "/ZKDTSMRoot"; + String workingPath = "/" + conf.get(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH, + ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH_DEAFULT) + "/ZKDTSMRoot"; - // Check if the created NameSpace exists. - Stat stat = curatorFramework.checkExists().forPath(workingPath); - assertNotNull(stat); + // Check if the created NameSpace exists. + Stat stat = curatorFramework.checkExists().forPath(workingPath); + assertNotNull(stat); - tm1.destroy(); - curatorFramework.close(); + tm1.destroy(); + } finally { + // Restore the default curator so later tests do not see a closed client. + ZKDelegationTokenSecretManager.setCurator(null); + curatorFramework.close(); + } } @Test @@ -616,36 +628,40 @@ public void testMultipleInit() throws Exception { DelegationTokenManager tm1 = new DelegationTokenManager(conf, new Text("foo")); DelegationTokenManager tm2 = new DelegationTokenManager(conf, new Text("bar")); - // When the init method is called, - // the ZKDelegationTokenSecretManager#startThread method will be called, - // and the creatingParentContainersIfNeeded will be called to create the nameSpace. - ExecutorService executorService = Executors.newFixedThreadPool(2); - - Callable tm1Callable = () -> { - tm1.init(); - return true; - }; - Callable tm2Callable = () -> { - tm2.init(); - return true; - }; - List> futures = executorService.invokeAll( - Arrays.asList(tm1Callable, tm2Callable)); - for(Future future : futures) { - assertTrue(future.get()); + try { + // When the init method is called, + // the ZKDelegationTokenSecretManager#startThread method will be called, + // and the creatingParentContainersIfNeeded will be called to create the nameSpace. + ExecutorService executorService = Executors.newFixedThreadPool(2); + + Callable tm1Callable = () -> { + tm1.init(); + return true; + }; + Callable tm2Callable = () -> { + tm2.init(); + return true; + }; + List> futures = executorService.invokeAll( + Arrays.asList(tm1Callable, tm2Callable)); + for (Future future : futures) { + assertTrue(future.get()); + } + executorService.shutdownNow(); + assertTrue(executorService.awaitTermination(1, TimeUnit.SECONDS)); + tm1.destroy(); + tm2.destroy(); + + String workingPath = "/" + conf.get(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH, + ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH_DEAFULT) + "/ZKDTSMRoot"; + + // Check if the created NameSpace exists. + Stat stat = curatorFramework.checkExists().forPath(workingPath); + assertNotNull(stat); + } finally { + // Restore the default curator so later tests do not see a closed client. + ZKDelegationTokenSecretManager.setCurator(null); + curatorFramework.close(); } - executorService.shutdownNow(); - assertTrue(executorService.awaitTermination(1, TimeUnit.SECONDS)); - tm1.destroy(); - tm2.destroy(); - - String workingPath = "/" + conf.get(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH, - ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH_DEAFULT) + "/ZKDTSMRoot"; - - // Check if the created NameSpace exists. - Stat stat = curatorFramework.checkExists().forPath(workingPath); - assertNotNull(stat); - - curatorFramework.close(); } } From d2bddcc76011e1fca5b1f39235bd5808ce8618fc Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Tue, 18 Aug 2026 21:39:01 +0800 Subject: [PATCH 2/3] HADOOP-19966. Ensure test cleanup runs on failure paths Move token manager destruction and executor shutdown into finally blocks in testCreatingParentContainersIfNeeded and testMultipleInit. Assisted-by: deepseek-v4-pro --- .../TestZKDelegationTokenSecretManager.java | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java index 9d70e54d6d24ec..df37aa3d96b308 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java @@ -558,8 +558,9 @@ public void testCreatingParentContainersIfNeeded() throws Exception { .build(); curatorFramework.start(); ZKDelegationTokenSecretManager.setCurator(curatorFramework); + DelegationTokenManager tm1 = null; try { - DelegationTokenManager tm1 = new DelegationTokenManager(conf, new Text("foo")); + tm1 = new DelegationTokenManager(conf, new Text("foo")); // When the init method is called, // the ZKDelegationTokenSecretManager#startThread method will be called, @@ -572,9 +573,10 @@ public void testCreatingParentContainersIfNeeded() throws Exception { // Check if the created NameSpace exists. Stat stat = curatorFramework.checkExists().forPath(workingPath); assertNotNull(stat); - - tm1.destroy(); } finally { + if (tm1 != null) { + tm1.destroy(); + } // Restore the default curator so later tests do not see a closed client. ZKDelegationTokenSecretManager.setCurator(null); curatorFramework.close(); @@ -628,11 +630,12 @@ public void testMultipleInit() throws Exception { DelegationTokenManager tm1 = new DelegationTokenManager(conf, new Text("foo")); DelegationTokenManager tm2 = new DelegationTokenManager(conf, new Text("bar")); + ExecutorService executorService = null; try { // When the init method is called, // the ZKDelegationTokenSecretManager#startThread method will be called, // and the creatingParentContainersIfNeeded will be called to create the nameSpace. - ExecutorService executorService = Executors.newFixedThreadPool(2); + executorService = Executors.newFixedThreadPool(2); Callable tm1Callable = () -> { tm1.init(); @@ -647,10 +650,6 @@ public void testMultipleInit() throws Exception { for (Future future : futures) { assertTrue(future.get()); } - executorService.shutdownNow(); - assertTrue(executorService.awaitTermination(1, TimeUnit.SECONDS)); - tm1.destroy(); - tm2.destroy(); String workingPath = "/" + conf.get(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH, ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH_DEAFULT) + "/ZKDTSMRoot"; @@ -659,6 +658,12 @@ public void testMultipleInit() throws Exception { Stat stat = curatorFramework.checkExists().forPath(workingPath); assertNotNull(stat); } finally { + if (executorService != null) { + executorService.shutdownNow(); + executorService.awaitTermination(1, TimeUnit.SECONDS); + } + tm1.destroy(); + tm2.destroy(); // Restore the default curator so later tests do not see a closed client. ZKDelegationTokenSecretManager.setCurator(null); curatorFramework.close(); From f35a83af9c70f5af3d89b141d07974fee822b7fe Mon Sep 17 00:00:00 2001 From: Cheng Pan Date: Wed, 2 Sep 2026 17:27:59 +0800 Subject: [PATCH 3/3] HADOOP-19966. Make cache init timeout configurable and release resources on failed start Add zk-dt-secret-manager.zkCacheInitTimeout (ms, default 0 waits indefinitely), call stopThreads() when startThreads() fails, and restore the immediate assertion in testNodesLoadedAfterRestart. Assisted-by: Claude Fable 5.1 --- .../ZKDelegationTokenSecretManager.java | 43 ++++++++++++++++--- .../TestZKDelegationTokenSecretManager.java | 8 +--- 2 files changed, 37 insertions(+), 14 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java index 65068bc76128d2..2de1c593d5284d 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/ZKDelegationTokenSecretManager.java @@ -48,6 +48,7 @@ import org.apache.hadoop.security.authentication.util.ZookeeperClient; import org.apache.hadoop.security.token.Token; import org.apache.hadoop.security.token.delegation.web.DelegationTokenManager; +import org.apache.hadoop.util.Preconditions; import static org.apache.hadoop.security.SecurityUtil.getServerPrincipal; import static org.apache.hadoop.util.Time.now; @@ -81,6 +82,10 @@ public abstract class ZKDelegationTokenSecretManager 0) { + if (!initialized.await(cacheInitTimeoutMs, TimeUnit.MILLISECONDS)) { + throw new IOException("Timed out after " + cacheInitTimeoutMs + + " ms waiting for " + cacheName + " cache initialization, " + + "consider increasing " + ZK_DTSM_ZK_CACHE_INIT_TIMEOUT); + } + } else { + initialized.await(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -515,8 +544,8 @@ public void stopThreads() { private void createPersistentNode(String nodePath) throws Exception { try { - // Pass empty data explicitly; Curator 5.2.0+ defaults data-less creates - // to the local address, which breaks parsing of the container nodes. + // Store empty data instead of Curator's default (the local address); + // HADOOP-17835 (3.4.0): CuratorCache includes the root node in the cache. zkClient.create().withMode(CreateMode.PERSISTENT).forPath(nodePath, new byte[0]); } catch (KeeperException.NodeExistsException ne) { LOG.debug(nodePath + " znode already exists !!"); diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java index df37aa3d96b308..4abc11d8f303e8 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/token/delegation/TestZKDelegationTokenSecretManager.java @@ -105,6 +105,7 @@ protected Configuration getSecretConf(String connectString) { conf.set(ZKDelegationTokenSecretManager.ZK_DTSM_ZNODE_WORKING_PATH, "testPath"); conf.set(ZKDelegationTokenSecretManager.ZK_DTSM_ZK_AUTH_TYPE, "none"); conf.setLong(ZKDelegationTokenSecretManager.ZK_DTSM_ZK_SHUTDOWN_TIMEOUT, 100); + conf.setLong(ZKDelegationTokenSecretManager.ZK_DTSM_ZK_CACHE_INIT_TIMEOUT, 10000); conf.setLong(DelegationTokenManager.UPDATE_INTERVAL, DAY_IN_SECS); conf.setLong(DelegationTokenManager.MAX_LIFETIME, DAY_IN_SECS); conf.setLong(DelegationTokenManager.RENEW_INTERVAL, DAY_IN_SECS); @@ -522,13 +523,6 @@ public Boolean get() { // The good token should be loaded on startup, and removed after expiry. id = smNew.decodeTokenIdentifier(token); - final AbstractDelegationTokenIdentifier idGood = id; - GenericTestUtils.waitFor(new Supplier() { - @Override - public Boolean get() { - return zksmNew.getTokenInfoFromMemory(idGood) != null; - } - }, 100, 5000); dtinfo = zksmNew.getTokenInfoFromMemory(id); assertNotNull(dtinfo, "good dt should be in memory!");