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

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

llap-client/src/test/org/apache/hadoop/hive/llap/registry/impl/TestLlapZookeeperRegistryImpl.java

@@ -20,25 +20,39 @@
 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;
 import org.junit.Test;
 
 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<LlapServiceInstance> serviceInstanceSet =
+              underTest.getInstances("LLAP", 10000);
+
+      // Then
+      Collection<LlapServiceInstance> 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<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
+    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<ACL> 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

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