HIVE-29802: Handle ClusterNotReadyException in ProactiveEviction and add Kerberos-aware tests

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);
+  }
+
+}

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) {

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 @@
         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<LlapServiceInstance> serviceInstanceSet =
+            underTest.getInstances("LLAP", 10000);
+
+    // Then
+    Collection<LlapServiceInstance> 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<ACL> 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

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);
     }

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<UserGroupInformation> 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;
+  }
+
+}