From 3620bbe2ff3a9f50a99b2445e82edeec223ffefb Mon Sep 17 00:00:00 2001 From: Gergely Farkas Date: Tue, 4 Aug 2026 16:04:44 +0200 Subject: [PATCH 1/4] HIVE-29802: Handle ClusterNotReadyException in ProactiveEviction and add Kerberos-aware tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When no LLAP daemons have started, the ZK paths don't exist yet. The PathChildrenCache attempts to create them using CREATOR_ALL_ACL, which requires an authenticated identity. Since HS2 is unauthenticated, ZooKeeper rejects this with InvalidACLException, surfacing as ClusterNotReadyException. Previously, this exception propagated as a RuntimeException from evict(), causing DDL operations like DROP DATABASE to fail when LLAP hadn't started. Fix: catch ClusterNotReadyException in ProactiveEviction.evict() and return silently — if no daemons are registered, there's nothing cached to evict. Co-Authored-By: Claude Opus 4.6 --- .../registry/ClusterNotReadyException.java | 28 +++ .../hive/registry/impl/ZkRegistryBase.java | 5 +- .../impl/TestLlapZookeeperRegistryImpl.java | 69 +++++++ .../hadoop/hive/llap/ProactiveEviction.java | 3 + .../hive/llap/TestProactiveEviction.java | 189 ++++++++++++++++++ 5 files changed, 292 insertions(+), 2 deletions(-) create mode 100644 llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java create mode 100644 ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java new file mode 100644 index 000000000000..72fa728b4653 --- /dev/null +++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.registry; + +import java.io.IOException; + +public class ClusterNotReadyException extends IOException { + + public ClusterNotReadyException(Throwable cause) { + super(cause); + } + +} diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java index 6e6ce31d6ff8..adebbcfc6fd0 100644 --- a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java +++ b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java @@ -48,6 +48,7 @@ import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.llap.LlapUtil; import org.apache.hadoop.hive.metastore.utils.SecurityUtils; +import org.apache.hadoop.hive.registry.ClusterNotReadyException; import org.apache.hadoop.hive.registry.RegistryUtilities; import org.apache.hadoop.hive.registry.ServiceInstance; import org.apache.hadoop.hive.registry.ServiceInstanceStateChangeListener; @@ -651,14 +652,14 @@ protected final synchronized PathChildrenCache ensureInstancesCache( long elapsedNs = System.nanoTime() - startTimeNs; if (deltaNs == 0 || deltaNs <= elapsedNs) { LOG.error("Unable to start curator PathChildrenCache", e); - throw new IOException(e); + throw new ClusterNotReadyException(e); } LOG.warn("The cluster is not started yet (InvalidACL); will retry"); try { Thread.sleep(Math.min(sleepTimeMs, (deltaNs - elapsedNs)/1000000L)); } catch (InterruptedException e1) { LOG.error("Interrupted while retrying the PathChildrenCache startup"); - throw new IOException(e1); + throw new ClusterNotReadyException(e1); } sleepTimeMs = sleepTimeMs << 1; } catch (Exception e) { diff --git a/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java b/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java index 7b227d4356bc..694d87f5100d 100644 --- a/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java +++ b/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java @@ -20,25 +20,34 @@ import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.api.ACLProvider; import org.apache.curator.retry.RetryOneTime; import org.apache.curator.test.TestingServer; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.llap.registry.LlapServiceInstance; +import org.apache.hadoop.hive.registry.ClusterNotReadyException; import org.apache.hadoop.hive.registry.ServiceInstanceSet; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.data.ACL; import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; import java.io.IOException; import java.lang.reflect.Field; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import static java.lang.Integer.parseInt; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; public class TestLlapZookeeperRegistryImpl { @@ -99,6 +108,66 @@ public void testRegister() throws Exception { parseInt(attributes.get(LlapRegistryService.LLAP_DAEMON_NUM_ENABLED_EXECUTORS))); } + @Test + public void testRetryOnInvalidACLException() throws Exception { + // Given + LlapZookeeperRegistryImpl underTest = + new LlapZookeeperRegistryImpl("ClientRegistryRetryTest", hiveConf); + + ACLProvider aclProvider = Mockito.mock(ACLProvider.class); + ACL allowAll = new ACL(ZooDefs.Perms.ALL, ZooDefs.Ids.ANYONE_ID_UNSAFE); + Mockito.when(aclProvider.getAclForPath(Mockito.any())). + thenReturn(Collections.emptyList()). // causes InvalidACLException + thenReturn(Collections.singletonList(allowAll)); // allow all + + CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory. + builder(). + connectString(server.getConnectString()). + sessionTimeoutMs(10000). + retryPolicy(new RetryOneTime(1000)). + aclProvider(aclProvider). + build(); + + trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); + underTest.start(); + + // When + ServiceInstanceSet serviceInstanceSet = + underTest.getInstances("LLAP", 10000); + + // Then + Collection llaps = serviceInstanceSet.getAll(); + assertEquals(0, llaps.size()); + Mockito.verify(aclProvider, Mockito.atLeast(4)).getAclForPath(Mockito.any()); + } + + @Test + public void testClusterNotReadyExceptionIsThrownWhenZkNodeNotExists() throws Exception { + // Given + LlapZookeeperRegistryImpl underTest = + new LlapZookeeperRegistryImpl("ClientRegistryClusterNotReadyTest", hiveConf); + + ACLProvider aclProvider = Mockito.mock(ACLProvider.class); + List secureAcls = new ArrayList<>(); + secureAcls.addAll(ZooDefs.Ids.READ_ACL_UNSAFE); // Read all to the world + secureAcls.addAll(ZooDefs.Ids.CREATOR_ALL_ACL); // Create/Delete/Write/Admin to creator + Mockito.when(aclProvider.getAclForPath(Mockito.any())).thenReturn(secureAcls); + CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory. + builder(). + connectString(server.getConnectString()). + sessionTimeoutMs(10000). + retryPolicy(new RetryOneTime(1000)). + aclProvider(aclProvider). + build(); + + trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); + underTest.start(); + + // When - Then + assertThrows(ClusterNotReadyException.class, + () -> underTest.getInstances("LLAP", 0)); + } + @Test public void testUpdate() throws Exception { // Given diff --git a/ql/src/java/org/apache/hadoop/hive/llap/ProactiveEviction.java b/ql/src/java/org/apache/hadoop/hive/llap/ProactiveEviction.java index b1fcf31a28fb..2d4a65929f29 100644 --- a/ql/src/java/org/apache/hadoop/hive/llap/ProactiveEviction.java +++ b/ql/src/java/org/apache/hadoop/hive/llap/ProactiveEviction.java @@ -39,6 +39,7 @@ import org.apache.hadoop.hive.llap.registry.LlapServiceInstance; import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService; import org.apache.hadoop.hive.metastore.Warehouse; +import org.apache.hadoop.hive.registry.ClusterNotReadyException; import org.apache.hadoop.io.retry.RetryPolicies; import org.apache.hadoop.io.retry.RetryPolicy; import org.apache.hadoop.net.NetUtils; @@ -100,6 +101,8 @@ public static void evict(Configuration conf, Request request) { EXECUTOR.execute(task); } + } catch (ClusterNotReadyException e) { + LOG.debug("LLAP cluster not ready, skipping proactive eviction.", e); } catch (IOException e) { throw new RuntimeException(e); } diff --git a/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java new file mode 100644 index 000000000000..65446f16be25 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java @@ -0,0 +1,189 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.llap; + +import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode; +import org.apache.curator.retry.RetryOneTime; +import org.apache.curator.test.TestingServer; +import org.apache.curator.utils.CloseableUtils; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService; +import org.apache.hadoop.hive.llap.registry.impl.LlapZookeeperRegistryImpl; +import org.apache.hadoop.hive.registry.impl.ZkRegistryBase; +import org.apache.hadoop.registry.client.binding.RegistryTypeUtils; +import org.apache.hadoop.registry.client.binding.RegistryUtils; +import org.apache.hadoop.registry.client.types.ServiceRecord; +import org.apache.hadoop.security.UserGroupInformation; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.URI; +import java.util.Map; +import java.util.concurrent.TimeUnit; + + +import static org.junit.Assert.fail; + +/** + * Tests for {@link ProactiveEviction} focusing on the ZooKeeper-based LLAP registry interaction + * with Kerberos authentication enabled. + * + * The tests use a local TestingServer (embedded ZooKeeper) and mock UGI to simulate a secure + * environment without requiring a real KDC. The "llap-sasl" namespace is used because + * HIVE_ZOOKEEPER_USE_KERBEROS is enabled, which is the namespace the registry uses in production + * when Kerberos is active. + */ +public class TestProactiveEviction { + + private HiveConf hiveConf = new HiveConf(); + + private CuratorFramework curatorFramework; + private TestingServer server; + + private UserGroupInformation ugi; + + MockedStatic userGroupInformationMockedStatic; + + @Before + public void setUp() throws Exception { + ugi = Mockito.mock(UserGroupInformation.class); + userGroupInformationMockedStatic = Mockito.mockStatic(UserGroupInformation.class); + userGroupInformationMockedStatic.when(UserGroupInformation::isSecurityEnabled).thenReturn(true); + userGroupInformationMockedStatic.when(UserGroupInformation::getCurrentUser).thenReturn(ugi); + Mockito.when(ugi.getShortUserName()).thenReturn("hive"); + + server = new TestingServer(); + server.start(); + + hiveConf.setVar(HiveConf.ConfVars.LLAP_DAEMON_SERVICE_HOSTS, "@testinstance"); + hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_USE_KERBEROS, true); + hiveConf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_QUORUM, server.getConnectString()); + hiveConf.setVar(HiveConf.ConfVars.HIVE_SERVER2_ZOOKEEPER_NAMESPACE, "testinstance"); + hiveConf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_NAMESPACE, "testinstance"); + hiveConf.setVar(HiveConf.ConfVars.LLAP_ZK_REGISTRY_USER, "hive"); + hiveConf.setVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_SESSION_TIMEOUT, "1000ms"); + hiveConf.setVar(HiveConf.ConfVars.LLAP_KERBEROS_PRINCIPAL, "hive/host@REALM"); + hiveConf.setVar(HiveConf.ConfVars.LLAP_KERBEROS_KEYTAB_FILE, "/keytab"); + } + + @After + public void tearDown() throws IOException { + server.stop(); + userGroupInformationMockedStatic.close(); + } + + /** + * Verifies that ProactiveEviction.evict() handles gracefully the case where Kerberos is enabled + * but no LLAP daemon instances are registered in ZooKeeper. The eviction should be skipped + * without throwing an exception; ClusterNotReadyException is caught internally. + */ + @Test + public void testEvictWithKerberosWithoutComputeInstances() throws Exception { + LlapProxy.setDaemon(true); + + ((Map) FieldUtils.readStaticField(LlapRegistryService.class, "yarnRegistries", true)).clear(); + + ProactiveEviction.Request.Builder llapEvictRequestBuilder = + ProactiveEviction.Request.Builder.create(); + llapEvictRequestBuilder.addTable("testDb", "testTable"); + ProactiveEviction.evict(hiveConf, llapEvictRequestBuilder.build()); + } + + /** + * Verifies that ProactiveEviction.evict() can discover and send eviction requests to LLAP + * daemon instances registered in ZooKeeper, with Kerberos enabled. + * + * The test pre-creates ZK znodes simulating LLAP daemons, then calls evict() which internally + * creates a fresh LlapRegistryService client that discovers them via the PathChildrenCache. + * The eviction tasks are fire-and-forget (they will fail to connect to the fake endpoints, + * but that's logged and swallowed by EvictionRequestTask). + */ + @Test + public void testEvictWithKerberosAndRegisteredComputes() throws Exception { + LlapProxy.setDaemon(true); + + String instanceName = "testinstance"; + + LlapZookeeperRegistryImpl registry = + new LlapZookeeperRegistryImpl(instanceName, hiveConf); + + curatorFramework = CuratorFrameworkFactory.builder() + .connectString(server.getConnectString()) + .sessionTimeoutMs(1000) + .namespace("llap-sasl") + .retryPolicy(new RetryOneTime(1000)) + .build(); + curatorFramework.start(); + + FieldUtils.writeField(registry, "zooKeeperClient", curatorFramework, true); + + String workersPath = (String) FieldUtils.readField(registry, "workersPath", true); + + PersistentEphemeralNode znode1 = createZnode(workersPath, "instance-1"); + PersistentEphemeralNode znode2 = createZnode(workersPath, "instance-2"); + + ((Map) FieldUtils.readStaticField(LlapRegistryService.class, "yarnRegistries", true)).clear(); + + ProactiveEviction.Request.Builder llapEvictRequestBuilder = + ProactiveEviction.Request.Builder.create(); + llapEvictRequestBuilder.addTable("testDb", "testTable"); + ProactiveEviction.evict(hiveConf, llapEvictRequestBuilder.build()); + + CloseableUtils.closeQuietly(znode1); + CloseableUtils.closeQuietly(znode2); + curatorFramework.close(); + } + + private PersistentEphemeralNode createZnode(String workersPath, String id) throws Exception { + ServiceRecord record = new ServiceRecord(); + record.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("llap", new InetSocketAddress("localhost", 4000))); + record.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("shuffle", new InetSocketAddress("localhost", 4001))); + record.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("llapmng", new InetSocketAddress("localhost", 4002))); + record.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("llapoutputformat", new InetSocketAddress("localhost", 4003))); + record.addExternalEndpoint( + RegistryTypeUtils.webEndpoint("services", new URI("http://localhost:4004"))); + record.set(LlapRegistryService.LLAP_DAEMON_NUM_ENABLED_EXECUTORS, "10"); + record.set(HiveConf.ConfVars.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB.varname, "100"); + record.set(ZkRegistryBase.UNIQUE_IDENTIFIER, id); + + PersistentEphemeralNode znode = new PersistentEphemeralNode( + curatorFramework, + PersistentEphemeralNode.Mode.EPHEMERAL_SEQUENTIAL, + workersPath + "/worker-", + new RegistryUtils.ServiceRecordMarshal().toBytes(record)); + znode.start(); + if (!znode.waitForInitialCreate(10, TimeUnit.SECONDS)) { + fail("Max znode creation wait time exhausted"); + } + return znode; + } + +} From 5c75838acf6fe466033caac52968562dffb32ce5 Mon Sep 17 00:00:00 2001 From: Gergely Farkas Date: Wed, 5 Aug 2026 15:25:44 +0200 Subject: [PATCH 2/4] HIVE-29802: Address review feedback - Restore thread interrupt flag with Thread.currentThread().interrupt() when catching InterruptedException in ZkRegistryBase, so higher-level cancellation/shutdown logic can observe the interrupt. - Include the exception in the LOG.error call so the cause is not lost. - Remove redundant TestingServer.start() call since the default constructor auto-starts the server. - Make test cleanup robust: close curatorFramework and TestingServer via CloseableUtils.closeQuietly() in tearDown with null guards, ensuring resources are released even if setup partially fails or a test throws. - Wrap CuratorFramework usage in TestLlapZookeeperRegistryImpl tests with try/finally to close the client and prevent thread/socket leaks. - Relax Mockito verification from atLeast(4) to atLeast(2) since the exact call count varies across Curator/ZK versions. - Rename testClusterNotReadyExceptionIsThrownWhenZkNodeNotExists to testClusterNotReadyExceptionOnImmediateTimeoutWithSecureAcl to accurately reflect that it tests the immediate-failure path (timeout=0). - Add testClusterNotReadyExceptionAfterRetriesWithSecureAcl which uses a small positive timeout (100ms) and asserts that retries occur before the deadline is reached, covering the retry-until-exhausted path. - Add serialVersionUID to ClusterNotReadyException and additional constructors accepting message and message+cause for more descriptive upstream logging without relying solely on nested exception text. - Add assertion in testEvictWithKerberosAndRegisteredComputes verifying that 2 LLAP instances are discovered before eviction is triggered. - Use static imports for Mockito.mock, when, verify, atLeast, any in TestLlapZookeeperRegistryImpl for cleaner test code. Co-Authored-By: Claude Opus 4.6 --- .../registry/ClusterNotReadyException.java | 10 ++ .../hive/registry/impl/ZkRegistryBase.java | 3 +- .../impl/TestLlapZookeeperRegistryImpl.java | 131 ++++++++++++------ .../hive/llap/TestProactiveEviction.java | 42 ++++-- 4 files changed, 131 insertions(+), 55 deletions(-) diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java index 72fa728b4653..bb3ae7bca517 100644 --- a/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java +++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java @@ -21,8 +21,18 @@ public class ClusterNotReadyException extends IOException { + private static final long serialVersionUID = 1L; + + public ClusterNotReadyException(String message) { + super(message); + } + public ClusterNotReadyException(Throwable cause) { super(cause); } + public ClusterNotReadyException(String message, Throwable cause) { + super(message, cause); + } + } diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java index adebbcfc6fd0..e0be21ac93d7 100644 --- a/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java +++ b/llap-client/src/java/org/apache/hadoop/hive/registry/impl/ZkRegistryBase.java @@ -658,7 +658,8 @@ protected final synchronized PathChildrenCache ensureInstancesCache( try { Thread.sleep(Math.min(sleepTimeMs, (deltaNs - elapsedNs)/1000000L)); } catch (InterruptedException e1) { - LOG.error("Interrupted while retrying the PathChildrenCache startup"); + Thread.currentThread().interrupt(); + LOG.error("Interrupted while retrying the PathChildrenCache startup", e1); throw new ClusterNotReadyException(e1); } sleepTimeMs = sleepTimeMs << 1; diff --git a/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java b/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java index 694d87f5100d..e75a62230f13 100644 --- a/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java +++ b/llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java @@ -23,6 +23,7 @@ import org.apache.curator.framework.api.ACLProvider; import org.apache.curator.retry.RetryOneTime; import org.apache.curator.test.TestingServer; +import org.apache.curator.utils.CloseableUtils; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.llap.registry.LlapServiceInstance; import org.apache.hadoop.hive.registry.ClusterNotReadyException; @@ -33,7 +34,6 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; -import org.mockito.Mockito; import java.io.IOException; import java.lang.reflect.Field; @@ -48,6 +48,11 @@ import static java.lang.Integer.parseInt; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class TestLlapZookeeperRegistryImpl { @@ -114,58 +119,106 @@ public void testRetryOnInvalidACLException() throws Exception { LlapZookeeperRegistryImpl underTest = new LlapZookeeperRegistryImpl("ClientRegistryRetryTest", hiveConf); - ACLProvider aclProvider = Mockito.mock(ACLProvider.class); + ACLProvider aclProvider = mock(ACLProvider.class); ACL allowAll = new ACL(ZooDefs.Perms.ALL, ZooDefs.Ids.ANYONE_ID_UNSAFE); - Mockito.when(aclProvider.getAclForPath(Mockito.any())). - thenReturn(Collections.emptyList()). // causes InvalidACLException - thenReturn(Collections.singletonList(allowAll)); // allow all + when(aclProvider.getAclForPath(any())) + .thenReturn(Collections.emptyList()) // causes InvalidACLException + .thenReturn(Collections.singletonList(allowAll)); // allow all + + CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory + .builder() + .connectString(server.getConnectString()) + .sessionTimeoutMs(10000) + .retryPolicy(new RetryOneTime(1000)) + .aclProvider(aclProvider) + .build(); - CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory. - builder(). - connectString(server.getConnectString()). - sessionTimeoutMs(10000). - retryPolicy(new RetryOneTime(1000)). - aclProvider(aclProvider). - build(); - - trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); - underTest.start(); - - // When - ServiceInstanceSet serviceInstanceSet = - underTest.getInstances("LLAP", 10000); - - // Then - Collection llaps = serviceInstanceSet.getAll(); - assertEquals(0, llaps.size()); - Mockito.verify(aclProvider, Mockito.atLeast(4)).getAclForPath(Mockito.any()); + try { + trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); + underTest.start(); + + // When + ServiceInstanceSet serviceInstanceSet = + underTest.getInstances("LLAP", 10000); + + // Then + Collection llaps = serviceInstanceSet.getAll(); + assertEquals(0, llaps.size()); + verify(aclProvider, atLeast(2)).getAclForPath(any()); + } finally { + CloseableUtils.closeQuietly(curatorFrameworkWithAclProvider); + } } @Test - public void testClusterNotReadyExceptionIsThrownWhenZkNodeNotExists() throws Exception { + public void testClusterNotReadyExceptionOnImmediateTimeoutWithSecureAcl() throws Exception { // Given LlapZookeeperRegistryImpl underTest = new LlapZookeeperRegistryImpl("ClientRegistryClusterNotReadyTest", hiveConf); - ACLProvider aclProvider = Mockito.mock(ACLProvider.class); + ACLProvider aclProvider = mock(ACLProvider.class); List secureAcls = new ArrayList<>(); secureAcls.addAll(ZooDefs.Ids.READ_ACL_UNSAFE); // Read all to the world secureAcls.addAll(ZooDefs.Ids.CREATOR_ALL_ACL); // Create/Delete/Write/Admin to creator - Mockito.when(aclProvider.getAclForPath(Mockito.any())).thenReturn(secureAcls); - CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory. - builder(). - connectString(server.getConnectString()). - sessionTimeoutMs(10000). - retryPolicy(new RetryOneTime(1000)). - aclProvider(aclProvider). - build(); + when(aclProvider.getAclForPath(any())).thenReturn(secureAcls); + CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory + .builder() + .connectString(server.getConnectString()) + .sessionTimeoutMs(10000) + .retryPolicy(new RetryOneTime(1000)) + .aclProvider(aclProvider) + .build(); + + try { + trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); + underTest.start(); + + // When - Then + assertThrows(ClusterNotReadyException.class, + () -> underTest.getInstances("LLAP", 0)); + } finally { + CloseableUtils.closeQuietly(curatorFrameworkWithAclProvider); + } + } - trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); - underTest.start(); + @Test + public void testClusterNotReadyExceptionAfterRetriesWithSecureAcl() throws Exception { + // Given + LlapZookeeperRegistryImpl underTest = + new LlapZookeeperRegistryImpl("ClientRegistryRetryTimeoutTest", hiveConf); - // When - Then - assertThrows(ClusterNotReadyException.class, - () -> underTest.getInstances("LLAP", 0)); + ACLProvider aclProvider = mock(ACLProvider.class); + List secureAcls = new ArrayList<>(); + secureAcls.addAll(ZooDefs.Ids.READ_ACL_UNSAFE); + secureAcls.addAll(ZooDefs.Ids.CREATOR_ALL_ACL); + when(aclProvider.getAclForPath(any())).thenReturn(secureAcls); + CuratorFramework curatorFrameworkWithAclProvider = CuratorFrameworkFactory + .builder() + .connectString(server.getConnectString()) + .sessionTimeoutMs(10000) + .retryPolicy(new RetryOneTime(1000)) + .aclProvider(aclProvider) + .build(); + + try { + trySetMock(underTest, "zooKeeperClient", curatorFrameworkWithAclProvider); + underTest.start(); + + long startMs = System.currentTimeMillis(); + + // When - Then: with a 100ms timeout, the method should retry before giving up + assertThrows(ClusterNotReadyException.class, + () -> underTest.getInstances("LLAP", 100)); + + long elapsedMs = System.currentTimeMillis() - startMs; + // Verify that retries actually occurred (elapsed time >= initial sleep of 16ms) + Assert.assertTrue("Expected retries before timeout, but elapsed was " + elapsedMs + "ms", + elapsedMs >= 16); + // Verify getAclForPath was called multiple times (at least initial attempt + one retry) + verify(aclProvider, atLeast(2)).getAclForPath(any()); + } finally { + CloseableUtils.closeQuietly(curatorFrameworkWithAclProvider); + } } @Test diff --git a/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java index 65446f16be25..b77b93252707 100644 --- a/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java +++ b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java @@ -26,6 +26,7 @@ import org.apache.curator.utils.CloseableUtils; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.llap.io.api.LlapProxy; +import org.apache.hadoop.hive.llap.registry.LlapServiceInstance; import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService; import org.apache.hadoop.hive.llap.registry.impl.LlapZookeeperRegistryImpl; import org.apache.hadoop.hive.registry.impl.ZkRegistryBase; @@ -42,10 +43,11 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.net.URI; +import java.util.Collection; import java.util.Map; import java.util.concurrent.TimeUnit; - +import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; /** @@ -77,7 +79,6 @@ public void setUp() throws Exception { Mockito.when(ugi.getShortUserName()).thenReturn("hive"); server = new TestingServer(); - server.start(); hiveConf.setVar(HiveConf.ConfVars.LLAP_DAEMON_SERVICE_HOSTS, "@testinstance"); hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_ZOOKEEPER_USE_KERBEROS, true); @@ -92,8 +93,16 @@ public void setUp() throws Exception { @After public void tearDown() throws IOException { - server.stop(); - userGroupInformationMockedStatic.close(); + if (curatorFramework != null) { + CloseableUtils.closeQuietly(curatorFramework); + curatorFramework = null; + } + if (server != null) { + CloseableUtils.closeQuietly(server); + } + if (userGroupInformationMockedStatic != null) { + userGroupInformationMockedStatic.close(); + } } /** @@ -148,6 +157,10 @@ public void testEvictWithKerberosAndRegisteredComputes() throws Exception { ((Map) FieldUtils.readStaticField(LlapRegistryService.class, "yarnRegistries", true)).clear(); + // Verify that the registry discovers both registered instances + Collection instances = registry.getInstances("LLAP", 10000).getAll(); + assertEquals(2, instances.size()); + ProactiveEviction.Request.Builder llapEvictRequestBuilder = ProactiveEviction.Request.Builder.create(); llapEvictRequestBuilder.addTable("testDb", "testTable"); @@ -155,30 +168,29 @@ public void testEvictWithKerberosAndRegisteredComputes() throws Exception { CloseableUtils.closeQuietly(znode1); CloseableUtils.closeQuietly(znode2); - curatorFramework.close(); } private PersistentEphemeralNode createZnode(String workersPath, String id) throws Exception { - ServiceRecord record = new ServiceRecord(); - record.addInternalEndpoint( + ServiceRecord serviceRecord = new ServiceRecord(); + serviceRecord.addInternalEndpoint( RegistryTypeUtils.ipcEndpoint("llap", new InetSocketAddress("localhost", 4000))); - record.addInternalEndpoint( + serviceRecord.addInternalEndpoint( RegistryTypeUtils.ipcEndpoint("shuffle", new InetSocketAddress("localhost", 4001))); - record.addInternalEndpoint( + serviceRecord.addInternalEndpoint( RegistryTypeUtils.ipcEndpoint("llapmng", new InetSocketAddress("localhost", 4002))); - record.addInternalEndpoint( + serviceRecord.addInternalEndpoint( RegistryTypeUtils.ipcEndpoint("llapoutputformat", new InetSocketAddress("localhost", 4003))); - record.addExternalEndpoint( + serviceRecord.addExternalEndpoint( RegistryTypeUtils.webEndpoint("services", new URI("http://localhost:4004"))); - record.set(LlapRegistryService.LLAP_DAEMON_NUM_ENABLED_EXECUTORS, "10"); - record.set(HiveConf.ConfVars.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB.varname, "100"); - record.set(ZkRegistryBase.UNIQUE_IDENTIFIER, id); + serviceRecord.set(LlapRegistryService.LLAP_DAEMON_NUM_ENABLED_EXECUTORS, "10"); + serviceRecord.set(HiveConf.ConfVars.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB.varname, "100"); + serviceRecord.set(ZkRegistryBase.UNIQUE_IDENTIFIER, id); PersistentEphemeralNode znode = new PersistentEphemeralNode( curatorFramework, PersistentEphemeralNode.Mode.EPHEMERAL_SEQUENTIAL, workersPath + "/worker-", - new RegistryUtils.ServiceRecordMarshal().toBytes(record)); + new RegistryUtils.ServiceRecordMarshal().toBytes(serviceRecord)); znode.start(); if (!znode.waitForInitialCreate(10, TimeUnit.SECONDS)) { fail("Max znode creation wait time exhausted"); From 7a1de8117d44144a5ef2f89834efb0c9bc913788 Mon Sep 17 00:00:00 2001 From: Gergely Farkas Date: Thu, 6 Aug 2026 15:33:47 +0200 Subject: [PATCH 3/4] HIVE-29802: Address review feedback in TestProactiveEviction - Remove unnecessary throws IOException from tearDown since CloseableUtils.closeQuietly() does not throw checked exceptions. - Use static imports for Mockito.mock, mockStatic, and when for cleaner test code consistent with TestLlapZookeeperRegistryImpl. - Organize imports: java.* first, then third-party/project imports, then static imports grouped by package. - Add explicit assertion to testEvictWithKerberosWithoutComputeInstances: wrap evict() in try/catch and fail if any exception is thrown, making the "handles gracefully" contract explicit rather than implicit. Co-Authored-By: Claude Opus 4.6 --- .../hive/llap/TestProactiveEviction.java | 32 +++++++++++-------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java index b77b93252707..322dbeb56f54 100644 --- a/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java +++ b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java @@ -17,6 +17,12 @@ */ package org.apache.hadoop.hive.llap; +import java.net.InetSocketAddress; +import java.net.URI; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.TimeUnit; + import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; @@ -38,17 +44,12 @@ import org.junit.Before; import org.junit.Test; import org.mockito.MockedStatic; -import org.mockito.Mockito; - -import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.URI; -import java.util.Collection; -import java.util.Map; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; /** * Tests for {@link ProactiveEviction} focusing on the ZooKeeper-based LLAP registry interaction @@ -72,11 +73,11 @@ public class TestProactiveEviction { @Before public void setUp() throws Exception { - ugi = Mockito.mock(UserGroupInformation.class); - userGroupInformationMockedStatic = Mockito.mockStatic(UserGroupInformation.class); + ugi = mock(UserGroupInformation.class); + userGroupInformationMockedStatic = mockStatic(UserGroupInformation.class); userGroupInformationMockedStatic.when(UserGroupInformation::isSecurityEnabled).thenReturn(true); userGroupInformationMockedStatic.when(UserGroupInformation::getCurrentUser).thenReturn(ugi); - Mockito.when(ugi.getShortUserName()).thenReturn("hive"); + when(ugi.getShortUserName()).thenReturn("hive"); server = new TestingServer(); @@ -92,7 +93,7 @@ public void setUp() throws Exception { } @After - public void tearDown() throws IOException { + public void tearDown() { if (curatorFramework != null) { CloseableUtils.closeQuietly(curatorFramework); curatorFramework = null; @@ -119,7 +120,12 @@ public void testEvictWithKerberosWithoutComputeInstances() throws Exception { ProactiveEviction.Request.Builder llapEvictRequestBuilder = ProactiveEviction.Request.Builder.create(); llapEvictRequestBuilder.addTable("testDb", "testTable"); - ProactiveEviction.evict(hiveConf, llapEvictRequestBuilder.build()); + + try { + ProactiveEviction.evict(hiveConf, llapEvictRequestBuilder.build()); + } catch (Exception e) { + fail("Expected evict() to handle missing instances gracefully, but threw: " + e); + } } /** From f0c61285375d23f8e8605dda223fc43b3d9b4bcd Mon Sep 17 00:00:00 2001 From: Gergely Farkas Date: Fri, 7 Aug 2026 09:04:16 +0200 Subject: [PATCH 4/4] HIVE-29802: Fix license header indentation in ClusterNotReadyException Co-Authored-By: Claude Opus 4.6 --- .../apache/hadoop/hive/registry/ClusterNotReadyException.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java index bb3ae7bca517..998c65b7b9f5 100644 --- a/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java +++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java @@ -7,7 +7,7 @@ * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS,