Merge branch 'apache:trunk' into YARN-11577

BUILDING.txt

@@ -163,13 +163,13 @@ Maven build goals:
  YARN Application Timeline Service V2 build options:
 
    YARN Timeline Service v.2 chooses Apache HBase as the primary backing storage. The supported
-   versions of Apache HBase are 1.2.6 (default) and 2.0.0-beta1.
+   versions of Apache HBase are 1.7.1 (default) and 2.2.4.
 
-  * HBase 1.2.6 is used by default to build Hadoop. The official releases are ready to use if you
-    plan on running Timeline Service v2 with HBase 1.2.6.
+  * HBase 1.7.1 is used by default to build Hadoop. The official releases are ready to use if you
+    plan on running Timeline Service v2 with HBase 1.7.1.
 
-  * Use -Dhbase.profile=2.0 to build Hadoop with HBase 2.0.0-beta1. Provide this option if you plan
-    on running Timeline Service v2 with HBase 2.0.
+  * Use -Dhbase.profile=2.0 to build Hadoop with HBase 2.2.4. Provide this option if you plan
+    on running Timeline Service v2 with HBase 2.x.
 
 
  Snappy build options:

LICENSE-binary

@@ -337,7 +337,7 @@ org.apache.kerby:kerby-xdr:2.0.3
 org.apache.kerby:token-provider:2.0.3
 org.apache.solr:solr-solrj:8.11.2
 org.apache.yetus:audience-annotations:0.5.0
-org.apache.zookeeper:zookeeper:3.6.3
+org.apache.zookeeper:zookeeper:3.7.2
 org.codehaus.jettison:jettison:1.5.4
 org.eclipse.jetty:jetty-annotations:9.4.51.v20230217
 org.eclipse.jetty:jetty-http:9.4.51.v20230217
@@ -360,7 +360,7 @@ org.objenesis:objenesis:2.6
 org.xerial.snappy:snappy-java:1.1.10.4
 org.yaml:snakeyaml:2.0
 org.wildfly.openssl:wildfly-openssl:1.1.3.Final
-software.amazon.awssdk:bundle:jar:2.20.128
+software.amazon.awssdk:bundle:jar:2.20.160
 
 
 --------------------------------------------------------------------------------

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StoreStatisticNames.java

@@ -407,10 +407,6 @@ public final class StoreStatisticNames {
   public static final String MULTIPART_UPLOAD_LIST
       = "multipart_upload_list";
 
-  /** Probe for store region: {@value}. */
-  public static final String STORE_REGION_PROBE
-      = "store_region_probe";
-
   private StoreStatisticNames() {
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ActiveStandbyElector.java

@@ -29,8 +29,10 @@
 import org.apache.hadoop.HadoopIllegalArgumentException;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.util.ZKUtil.ZKAuthInfo;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Watcher;
@@ -48,6 +50,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.naming.ConfigurationException;
+
+import org.apache.hadoop.security.SecurityUtil.TruststoreKeystore;
+
 /**
  * 
  * This class implements a simple library to perform leader election on top of
@@ -170,6 +176,7 @@ enum State {
   private final int zkSessionTimeout;
   private final List<ACL> zkAcl;
   private final List<ZKAuthInfo> zkAuthInfo;
+  private TruststoreKeystore truststoreKeystore;
   private byte[] appData;
   private final String zkLockFilePath;
   private final String zkBreadCrumbPath;
@@ -209,6 +216,7 @@ enum State {
    * @param app
    *          reference to callback interface object
    * @param maxRetryNum maxRetryNum.
+   * @param truststoreKeystore truststore keystore, that we will use for ZK if SSL/TLS is enabled
    * @throws IOException raised on errors performing I/O.
    * @throws HadoopIllegalArgumentException
    *         if valid data is not supplied.
@@ -218,10 +226,10 @@ enum State {
   public ActiveStandbyElector(String zookeeperHostPorts,
       int zookeeperSessionTimeout, String parentZnodeName, List<ACL> acl,
       List<ZKAuthInfo> authInfo, ActiveStandbyElectorCallback app,
-      int maxRetryNum) throws IOException, HadoopIllegalArgumentException,
-      KeeperException {
+      int maxRetryNum, TruststoreKeystore truststoreKeystore)
+          throws IOException, HadoopIllegalArgumentException, KeeperException {
     this(zookeeperHostPorts, zookeeperSessionTimeout, parentZnodeName, acl,
-      authInfo, app, maxRetryNum, true);
+            authInfo, app, maxRetryNum, true, truststoreKeystore);
   }
 
   /**
@@ -254,6 +262,7 @@ public ActiveStandbyElector(String zookeeperHostPorts,
    * @param failFast
    *          whether need to add the retry when establishing ZK connection.
    * @param maxRetryNum max Retry Num
+   * @param truststoreKeystore truststore keystore, that we will use for ZK if SSL/TLS is enabled
    * @throws IOException
    *          raised on errors performing I/O.
    * @throws HadoopIllegalArgumentException
@@ -264,7 +273,7 @@ public ActiveStandbyElector(String zookeeperHostPorts,
   public ActiveStandbyElector(String zookeeperHostPorts,
       int zookeeperSessionTimeout, String parentZnodeName, List<ACL> acl,
       List<ZKAuthInfo> authInfo, ActiveStandbyElectorCallback app,
-      int maxRetryNum, boolean failFast) throws IOException,
+      int maxRetryNum, boolean failFast, TruststoreKeystore truststoreKeystore) throws IOException,
       HadoopIllegalArgumentException, KeeperException {
     if (app == null || acl == null || parentZnodeName == null
         || zookeeperHostPorts == null || zookeeperSessionTimeout <= 0) {
@@ -279,6 +288,7 @@ public ActiveStandbyElector(String zookeeperHostPorts,
     zkLockFilePath = znodeWorkingDir + "/" + LOCK_FILENAME;
     zkBreadCrumbPath = znodeWorkingDir + "/" + BREADCRUMB_FILENAME;
     this.maxRetryNum = maxRetryNum;
+    this.truststoreKeystore = truststoreKeystore;
 
     // establish the ZK Connection for future API calls
     if (failFast) {
@@ -740,7 +750,19 @@ protected synchronized ZooKeeper connectToZooKeeper() throws IOException,
    * @throws IOException raised on errors performing I/O.
    */
   protected ZooKeeper createZooKeeper() throws IOException {
-    return new ZooKeeper(zkHostPort, zkSessionTimeout, watcher);
+    ZKClientConfig zkClientConfig = new ZKClientConfig();
+    if (truststoreKeystore != null) {
+      try {
+        SecurityUtil.setSslConfiguration(zkClientConfig, truststoreKeystore);
+      } catch (ConfigurationException ce) {
+        throw new IOException(ce);
+      }
+    }
+    return initiateZookeeper(zkClientConfig);
+  }
+
+  protected ZooKeeper initiateZookeeper(ZKClientConfig zkClientConfig) throws IOException {
+    return new ZooKeeper(zkHostPort, zkSessionTimeout, watcher, zkClientConfig);
   }
 
   private void fatalError(String errorMessage) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ha/ZKFailoverController.java

@@ -59,6 +59,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.security.SecurityUtil.TruststoreKeystore;
+
 @InterfaceAudience.LimitedPrivate("HDFS")
 public abstract class ZKFailoverController {
 
@@ -147,6 +149,7 @@ protected abstract void checkRpcAdminAccess()
   protected abstract InetSocketAddress getRpcAddressToBindTo();
   protected abstract PolicyProvider getPolicyProvider();
   protected abstract List<HAServiceTarget> getAllOtherNodes();
+  protected abstract boolean isSSLEnabled();
 
   /**
    * Return the name of a znode inside the configured parent znode in which
@@ -372,9 +375,10 @@ private void initZK() throws HadoopIllegalArgumentException, IOException,
     int maxRetryNum = conf.getInt(
         CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_KEY,
         CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
+    TruststoreKeystore truststoreKeystore = isSSLEnabled() ? new TruststoreKeystore(conf) : null;
     elector = new ActiveStandbyElector(zkQuorum,
         zkTimeout, getParentZnode(), zkAcls, zkAuths,
-        new ElectorCallbacks(), maxRetryNum);
+        new ElectorCallbacks(), maxRetryNum, truststoreKeystore);
   }
 
   private String getParentZnode() {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java

@@ -18,6 +18,9 @@
 
 package org.apache.hadoop.ipc;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT;
+
 import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
@@ -63,7 +66,7 @@ static Class<? extends RpcScheduler> convertSchedulerClass(
   }
 
   private volatile boolean clientBackOffEnabled;
-  private boolean serverFailOverEnabled;
+  private volatile boolean serverFailOverEnabled;
 
   // Atomic refs point to active callQueue
   // We have two so we can better control swapping
@@ -81,18 +84,15 @@ public CallQueueManager(Class<? extends BlockingQueue<E>> backingClass,
         namespace, conf);
     int[] capacityWeights = parseCapacityWeights(priorityLevels,
         namespace, conf);
+    this.serverFailOverEnabled = getServerFailOverEnable(namespace, conf);
     BlockingQueue<E> bq = createCallQueueInstance(backingClass,
         priorityLevels, maxQueueSize, namespace, capacityWeights, conf);
     this.clientBackOffEnabled = clientBackOffEnabled;
-    this.serverFailOverEnabled = conf.getBoolean(
-        namespace + "." +
-        CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE,
-        CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT);
     this.putRef = new AtomicReference<BlockingQueue<E>>(bq);
     this.takeRef = new AtomicReference<BlockingQueue<E>>(bq);
     LOG.info("Using callQueue: {}, queueCapacity: {}, " +
-        "scheduler: {}, ipcBackoff: {}.",
-        backingClass, maxQueueSize, schedulerClass, clientBackOffEnabled);
+        "scheduler: {}, ipcBackoff: {}, ipcFailOver: {}.",
+        backingClass, maxQueueSize, schedulerClass, clientBackOffEnabled, serverFailOverEnabled);
   }
 
   @VisibleForTesting // only!
@@ -105,6 +105,41 @@ public CallQueueManager(Class<? extends BlockingQueue<E>> backingClass,
     this.serverFailOverEnabled = serverFailOverEnabled;
   }
 
+  /**
+   * Return boolean value configured by property 'ipc.<port>.callqueue.overflow.trigger.failover'
+   * if it is present. If the config is not present, default config
+   * (without port) is used to derive class i.e 'ipc.callqueue.overflow.trigger.failover',
+   * and derived value is returned if configured. Otherwise, default value
+   * {@link CommonConfigurationKeys#IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT} is returned.
+   *
+   * @param namespace Namespace "ipc" + "." + Server's listener port.
+   * @param conf Configuration properties.
+   * @return Value returned based on configuration.
+   */
+  private boolean getServerFailOverEnable(String namespace, Configuration conf) {
+    String propertyKey = namespace + "." +
+        CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE;
+
+    if (conf.get(propertyKey) != null) {
+      return conf.getBoolean(propertyKey,
+          CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT);
+    }
+
+    String[] nsPort = namespace.split("\\.");
+    if (nsPort.length == 2) {
+      // Only if ns is split with ".", we can separate namespace and port.
+      // In the absence of "ipc.<port>.callqueue.overflow.trigger.failover" property,
+      // we look up "ipc.callqueue.overflow.trigger.failover" property.
+      return conf.getBoolean(nsPort[0] + "."
+          + IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE, IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT);
+    }
+
+    // Otherwise return default value.
+    LOG.info("{} not specified set default value is {}",
+        IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE, IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT);
+    return CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT;
+  }
+
   private static <T extends RpcScheduler> T createScheduler(
       Class<T> theClass, int priorityLevels, String ns, Configuration conf) {
     // Used for custom, configurable scheduler
@@ -155,9 +190,9 @@ private <T extends BlockingQueue<E>> T createCallQueueInstance(
     // Used for custom, configurable callqueues
     try {
       Constructor<T> ctor = theClass.getDeclaredConstructor(int.class,
-          int.class, String.class, int[].class, Configuration.class);
-      return ctor.newInstance(priorityLevels, maxLen, ns,
-          capacityWeights, conf);
+          int.class, String.class, int[].class, boolean.class, Configuration.class);
+      return ctor.newInstance(priorityLevels, maxLen, ns, capacityWeights,
+          this.serverFailOverEnabled, conf);
     } catch (RuntimeException e) {
       throw e;
     } catch (InvocationTargetException e) {
@@ -199,6 +234,20 @@ boolean isClientBackoffEnabled() {
     return clientBackOffEnabled;
   }
 
+  @VisibleForTesting
+  public boolean isServerFailOverEnabled() {
+    return serverFailOverEnabled;
+  }
+
+  @VisibleForTesting
+  public boolean isServerFailOverEnabledByQueue() {
+    BlockingQueue<E> bq = putRef.get();
+    if (bq instanceof FairCallQueue) {
+      return ((FairCallQueue<E>) bq).isServerFailOverEnabled();
+    }
+    return false;
+  }
+
   // Based on policy to determine back off current call
   boolean shouldBackOff(Schedulable e) {
     return scheduler.shouldBackOff(e);
@@ -421,6 +470,9 @@ public synchronized void swapQueue(
     RpcScheduler newScheduler = createScheduler(schedulerClass, priorityLevels,
         ns, conf);
     int[] capacityWeights = parseCapacityWeights(priorityLevels, ns, conf);
+
+    // Update serverFailOverEnabled.
+    this.serverFailOverEnabled = getServerFailOverEnable(ns, conf);
     BlockingQueue<E> newQ = createCallQueueInstance(queueClassToUse,
         priorityLevels, maxSize, ns, capacityWeights, conf);
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/FairCallQueue.java

@@ -35,7 +35,6 @@
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.ipc.CallQueueManager.CallQueueOverflowException;
 import org.apache.hadoop.metrics2.MetricsCollector;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
@@ -90,7 +89,16 @@ private void signalNotEmpty() {
   public FairCallQueue(int priorityLevels, int capacity, String ns,
       Configuration conf) {
     this(priorityLevels, capacity, ns,
-        CallQueueManager.getDefaultQueueCapacityWeights(priorityLevels), conf);
+        CallQueueManager.getDefaultQueueCapacityWeights(priorityLevels),
+        false, conf);
+  }
+
+  @VisibleForTesting
+  public FairCallQueue(int priorityLevels, int capacity, String ns, boolean serverFailOverEnabled,
+      Configuration conf) {
+    this(priorityLevels, capacity, ns,
+        CallQueueManager.getDefaultQueueCapacityWeights(priorityLevels),
+        serverFailOverEnabled, conf);
   }
 
   /**
@@ -101,18 +109,21 @@ public FairCallQueue(int priorityLevels, int capacity, String ns,
    * @param ns the prefix to use for configuration
    * @param capacityWeights the weights array for capacity allocation
    *                        among subqueues
+   * @param serverFailOverEnabled whether or not to enable callqueue overflow trigger failover
+   *                              for stateless servers when RPC call queue is filled
    * @param conf the configuration to read from
    * Notes: Each sub-queue has a capacity of `capacity / numSubqueues`.
    * The first or the highest priority sub-queue has an excess capacity
    * of `capacity % numSubqueues`
    */
   public FairCallQueue(int priorityLevels, int capacity, String ns,
-      int[] capacityWeights, Configuration conf) {
+      int[] capacityWeights, boolean serverFailOverEnabled, Configuration conf) {
     if(priorityLevels < 1) {
       throw new IllegalArgumentException("Number of Priority Levels must be " +
           "at least 1");
     }
     int numQueues = priorityLevels;
+    this.serverFailOverEnabled = serverFailOverEnabled;
     LOG.info("FairCallQueue is in use with " + numQueues +
         " queues with total capacity of " + capacity);
 
@@ -135,10 +146,6 @@ public FairCallQueue(int priorityLevels, int capacity, String ns,
       }
       this.overflowedCalls.add(new AtomicLong(0));
     }
-    this.serverFailOverEnabled = conf.getBoolean(
-        ns + "." +
-        CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE,
-        CommonConfigurationKeys.IPC_CALLQUEUE_SERVER_FAILOVER_ENABLE_DEFAULT);
 
     this.multiplexer = new WeightedRoundRobinMultiplexer(numQueues, ns, conf);
     // Make this the active source of metrics
@@ -493,4 +500,9 @@ public long[] getOverflowedCalls() {
   public void setMultiplexer(RpcMultiplexer newMux) {
     this.multiplexer = newMux;
   }
+
+  @VisibleForTesting
+  public boolean isServerFailOverEnabled() {
+    return serverFailOverEnabled;
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -3857,6 +3857,16 @@ public void setClientBackoffEnabled(boolean value) {
     callQueue.setClientBackoffEnabled(value);
   }
 
+  @VisibleForTesting
+  public boolean isServerFailOverEnabled() {
+    return callQueue.isServerFailOverEnabled();
+  }
+
+  @VisibleForTesting
+  public boolean isServerFailOverEnabledByQueue() {
+    return callQueue.isServerFailOverEnabledByQueue();
+  }
+
   /**
    * The maximum size of the rpc call queue of this server.
    * @return The maximum size of the rpc call queue.