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..998c65b7b9f5 --- /dev/null +++ b/llap-client/src/java/org/apache/hadoop/hive/registry/ClusterNotReadyException.java @@ -0,0 +1,38 @@ +/* + * 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 { + + 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 6e6ce31d6ff8..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 @@ -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,15 @@ 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); + Thread.currentThread().interrupt(); + LOG.error("Interrupted while retrying the PathChildrenCache startup", 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..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 @@ -20,11 +20,16 @@ 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.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; 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; @@ -32,13 +37,22 @@ 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; +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 { @@ -99,6 +113,114 @@ 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 = mock(ACLProvider.class); + ACL allowAll = new ACL(ZooDefs.Perms.ALL, ZooDefs.Ids.ANYONE_ID_UNSAFE); + 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(); + + 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 testClusterNotReadyExceptionOnImmediateTimeoutWithSecureAcl() throws Exception { + // Given + LlapZookeeperRegistryImpl underTest = + new LlapZookeeperRegistryImpl("ClientRegistryClusterNotReadyTest", hiveConf); + + 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 + 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); + } + } + + @Test + public void testClusterNotReadyExceptionAfterRetriesWithSecureAcl() throws Exception { + // Given + LlapZookeeperRegistryImpl underTest = + new LlapZookeeperRegistryImpl("ClientRegistryRetryTimeoutTest", hiveConf); + + 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 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..322dbeb56f54 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/llap/TestProactiveEviction.java @@ -0,0 +1,207 @@ +/* + * 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 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; +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.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; +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 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 + * 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 = mock(UserGroupInformation.class); + userGroupInformationMockedStatic = mockStatic(UserGroupInformation.class); + userGroupInformationMockedStatic.when(UserGroupInformation::isSecurityEnabled).thenReturn(true); + userGroupInformationMockedStatic.when(UserGroupInformation::getCurrentUser).thenReturn(ugi); + when(ugi.getShortUserName()).thenReturn("hive"); + + server = new TestingServer(); + + 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() { + if (curatorFramework != null) { + CloseableUtils.closeQuietly(curatorFramework); + curatorFramework = null; + } + if (server != null) { + CloseableUtils.closeQuietly(server); + } + if (userGroupInformationMockedStatic != null) { + 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"); + + try { + ProactiveEviction.evict(hiveConf, llapEvictRequestBuilder.build()); + } catch (Exception e) { + fail("Expected evict() to handle missing instances gracefully, but threw: " + e); + } + } + + /** + * 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(); + + // 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"); + ProactiveEviction.evict(hiveConf, llapEvictRequestBuilder.build()); + + CloseableUtils.closeQuietly(znode1); + CloseableUtils.closeQuietly(znode2); + } + + private PersistentEphemeralNode createZnode(String workersPath, String id) throws Exception { + ServiceRecord serviceRecord = new ServiceRecord(); + serviceRecord.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("llap", new InetSocketAddress("localhost", 4000))); + serviceRecord.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("shuffle", new InetSocketAddress("localhost", 4001))); + serviceRecord.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("llapmng", new InetSocketAddress("localhost", 4002))); + serviceRecord.addInternalEndpoint( + RegistryTypeUtils.ipcEndpoint("llapoutputformat", new InetSocketAddress("localhost", 4003))); + serviceRecord.addExternalEndpoint( + RegistryTypeUtils.webEndpoint("services", new URI("http://localhost:4004"))); + 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(serviceRecord)); + znode.start(); + if (!znode.waitForInitialCreate(10, TimeUnit.SECONDS)) { + fail("Max znode creation wait time exhausted"); + } + return znode; + } + +}