Merge branch 'trunk' into HADOOP-18646

# Conflicts:
#	LICENSE-binary

Author: nao-it

LICENSE-binary

@@ -210,9 +210,9 @@ hadoop-hdfs-project/hadoop-hdfs/src/main/webapps/static/nvd3-1.8.5.* (css and js
 hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AbstractFuture.java
 hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/TimeoutFuture.java
 
-com.aliyun:aliyun-java-sdk-core:3.4.0
-com.aliyun:aliyun-java-sdk-ecs:4.2.0
-com.aliyun:aliyun-java-sdk-ram:3.0.0
+com.aliyun:aliyun-java-sdk-core:4.5.10
+com.aliyun:aliyun-java-sdk-kms:2.11.0
+com.aliyun:aliyun-java-sdk-ram:3.1.0
 com.aliyun:aliyun-java-sdk-sts:3.0.0
 com.aliyun.oss:aliyun-sdk-oss:3.13.2
 com.amazonaws:aws-java-sdk-bundle:1.12.316
@@ -259,36 +259,36 @@ io.grpc:grpc-netty:1.26.0
 io.grpc:grpc-protobuf:1.26.0
 io.grpc:grpc-protobuf-lite:1.26.0
 io.grpc:grpc-stub:1.26.0
-io.netty:netty-all:4.1.89.Final
-io.netty:netty-buffer:4.1.89.Final
-io.netty:netty-codec:4.1.89.Final
-io.netty:netty-codec-dns:4.1.89.Final
-io.netty:netty-codec-haproxy:4.1.89.Final
-io.netty:netty-codec-http:4.1.89.Final
-io.netty:netty-codec-http2:4.1.89.Final
-io.netty:netty-codec-memcache:4.1.89.Final
-io.netty:netty-codec-mqtt:4.1.89.Final
-io.netty:netty-codec-redis:4.1.89.Final
-io.netty:netty-codec-smtp:4.1.89.Final
-io.netty:netty-codec-socks:4.1.89.Final
-io.netty:netty-codec-stomp:4.1.89.Final
-io.netty:netty-codec-xml:4.1.89.Final
-io.netty:netty-common:4.1.89.Final
-io.netty:netty-handler:4.1.89.Final
-io.netty:netty-handler-proxy:4.1.89.Final
-io.netty:netty-resolver:4.1.89.Final
-io.netty:netty-resolver-dns:4.1.89.Final
-io.netty:netty-transport:4.1.89.Final
-io.netty:netty-transport-rxtx:4.1.89.Final
-io.netty:netty-transport-sctp:4.1.89.Final
-io.netty:netty-transport-udt:4.1.89.Final
-io.netty:netty-transport-classes-epoll:4.1.89.Final
-io.netty:netty-transport-native-unix-common:4.1.89.Final
-io.netty:netty-transport-classes-kqueue:4.1.89.Final
-io.netty:netty-resolver-dns-classes-macos:4.1.89.Final
-io.netty:netty-transport-native-epoll:4.1.89.Final
-io.netty:netty-transport-native-kqueue:4.1.89.Final
-io.netty:netty-resolver-dns-native-macos:4.1.89.Final
+io.netty:netty-all:4.1.77.Final
+io.netty:netty-buffer:4.1.77.Final
+io.netty:netty-codec:4.1.77.Final
+io.netty:netty-codec-dns:4.1.77.Final
+io.netty:netty-codec-haproxy:4.1.77.Final
+io.netty:netty-codec-http:4.1.77.Final
+io.netty:netty-codec-http2:4.1.77.Final
+io.netty:netty-codec-memcache:4.1.77.Final
+io.netty:netty-codec-mqtt:4.1.77.Final
+io.netty:netty-codec-redis:4.1.77.Final
+io.netty:netty-codec-smtp:4.1.77.Final
+io.netty:netty-codec-socks:4.1.77.Final
+io.netty:netty-codec-stomp:4.1.77.Final
+io.netty:netty-codec-xml:4.1.77.Final
+io.netty:netty-common:4.1.77.Final
+io.netty:netty-handler:4.1.77.Final
+io.netty:netty-handler-proxy:4.1.77.Final
+io.netty:netty-resolver:4.1.77.Final
+io.netty:netty-resolver-dns:4.1.77.Final
+io.netty:netty-transport:4.1.77.Final
+io.netty:netty-transport-rxtx:4.1.77.Final
+io.netty:netty-transport-sctp:4.1.77.Final
+io.netty:netty-transport-udt:4.1.77.Final
+io.netty:netty-transport-classes-epoll:4.1.77.Final
+io.netty:netty-transport-native-unix-common:4.1.77.Final
+io.netty:netty-transport-classes-kqueue:4.1.77.Final
+io.netty:netty-resolver-dns-classes-macos:4.1.77.Final
+io.netty:netty-transport-native-epoll:4.1.77.Final
+io.netty:netty-transport-native-kqueue:4.1.77.Final
+io.netty:netty-resolver-dns-native-macos:4.1.77.Final
 io.opencensus:opencensus-api:0.12.3
 io.opencensus:opencensus-contrib-grpc-metrics:0.12.3
 io.reactivex:rxjava:1.3.8
@@ -357,6 +357,9 @@ org.eclipse.jetty:jetty-xml:9.4.48.v20220622
 org.eclipse.jetty.websocket:javax-websocket-client-impl:9.4.48.v20220622
 org.eclipse.jetty.websocket:javax-websocket-server-impl:9.4.48.v20220622
 org.ehcache:ehcache:3.3.1
+org.ini4j:ini4j:0.5.4
+org.jetbrains.kotlin:kotlin-stdlib:1.4.10
+org.jetbrains.kotlin:kotlin-stdlib-common:1.4.10
 org.lz4:lz4-java:1.7.1
 org.objenesis:objenesis:2.6
 org.xerial.snappy:snappy-java:1.0.5
@@ -516,6 +519,8 @@ Eclipse Public License 1.0
 --------------------------
 
 junit:junit:4.13.2
+org.jacoco:org.jacoco.agent:0.8.5
+
 
 
 HSQL License

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SecurityUtil.java

@@ -314,7 +314,8 @@ public static void login(final Configuration conf,
 
     String keytabFilename = conf.get(keytabFileKey);
     if (keytabFilename == null || keytabFilename.length() == 0) {
-      throw new IOException("Running in secure mode, but config doesn't have a keytab");
+      throw new IOException(
+          "Running in secure mode, but config doesn't have a keytab for key: " + keytabFileKey);
     }
 
     String principalConfig = conf.get(userNameKey, System

hadoop-common-project/hadoop-common/src/site/markdown/Metrics.md

@@ -370,6 +370,9 @@ Each metrics record contains tags such as SessionId and Hostname as additional i
 |:---- |:---- |
 | `BytesWritten` | Total number of bytes written to DataNode |
 | `BytesRead` | Total number of bytes read from DataNode |
+| `ReadTransferRateNumOps` | Total number of data read transfers |
+| `ReadTransferRateAvgTime` | Average transfer rate of bytes read from DataNode, measured in bytes per second. |
+| `ReadTransferRate`*num*`s(50/75/90/95/99)thPercentileRate` | The 50/75/90/95/99th percentile of the transfer rate of bytes read from DataNode, measured in bytes per second. |
 | `BlocksWritten` | Total number of blocks written to DataNode |
 | `BlocksRead` | Total number of blocks read from DataNode |
 | `BlocksReplicated` | Total number of blocks replicated |

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/KMSConfiguration.java

@@ -20,7 +20,7 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.log4j.PropertyConfigurator;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -104,8 +104,6 @@ public class KMSConfiguration {
 
   public static final boolean KEY_AUTHORIZATION_ENABLE_DEFAULT = true;
 
-  private static final String LOG4J_PROPERTIES = "kms-log4j.properties";
-
   static {
     Configuration.addDefaultResource(KMS_DEFAULT_XML);
     Configuration.addDefaultResource(KMS_SITE_XML);
@@ -163,31 +161,20 @@ public static boolean isACLsFileNewer(long time) {
     return newer;
   }
 
-  public static void initLogging() {
-    String confDir = System.getProperty(KMS_CONFIG_DIR);
-    if (confDir == null) {
-      throw new RuntimeException("System property '" +
-          KMSConfiguration.KMS_CONFIG_DIR + "' not defined");
+  /**
+   * Validate whether "kms.config.dir" and "log4j.configuration" are defined in the System
+   * properties. If not, abort the KMS WebServer.
+   */
+  public static void validateSystemProps() {
+    if (System.getProperty(KMS_CONFIG_DIR) == null) {
+      String errorMsg = "System property '" + KMS_CONFIG_DIR + "' not defined";
+      System.err.println("Aborting KMSWebServer because " + errorMsg);
+      throw new RuntimeException(errorMsg);
     }
     if (System.getProperty("log4j.configuration") == null) {
-      System.setProperty("log4j.defaultInitOverride", "true");
-      boolean fromClasspath = true;
-      File log4jConf = new File(confDir, LOG4J_PROPERTIES).getAbsoluteFile();
-      if (log4jConf.exists()) {
-        PropertyConfigurator.configureAndWatch(log4jConf.getPath(), 1000);
-        fromClasspath = false;
-      } else {
-        ClassLoader cl = Thread.currentThread().getContextClassLoader();
-        URL log4jUrl = cl.getResource(LOG4J_PROPERTIES);
-        if (log4jUrl != null) {
-          PropertyConfigurator.configure(log4jUrl);
-        }
-      }
-      LOG.debug("KMS log starting");
-      if (fromClasspath) {
-        LOG.warn("Log4j configuration file '{}' not found", LOG4J_PROPERTIES);
-        LOG.warn("Logging with INFO level to standard output");
-      }
+      String errorMsg = "System property 'log4j.configuration' not defined";
+      System.err.println("Aborting KMSWebServer because " + errorMsg);
+      throw new RuntimeException(errorMsg);
     }
   }
 }

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/KMSWebServer.java

@@ -185,7 +185,7 @@ public URL getKMSUrl() {
   }
 
   public static void main(String[] args) throws Exception {
-    KMSConfiguration.initLogging();
+    KMSConfiguration.validateSystemProps();
     StringUtils.startupShutdownMessage(KMSWebServer.class, args, LOG);
     Configuration conf = KMSConfiguration.getKMSConf();
     Configuration sslConf = SSLFactory.readSSLConfiguration(conf, SSLFactory.Mode.SERVER);

hadoop-common-project/hadoop-kms/src/main/libexec/shellprofile.d/hadoop-kms.sh

@@ -49,6 +49,8 @@ function hadoop_subcommand_kms
     "-Dkms.config.dir=${HADOOP_CONF_DIR}"
   hadoop_add_param HADOOP_OPTS "-Dkms.log.dir=" \
     "-Dkms.log.dir=${HADOOP_LOG_DIR}"
+  hadoop_add_param HADOOP_OPTS "-Dlog4j.configuration=" \
+    "-Dlog4j.configuration=file:${HADOOP_CONF_DIR}/kms-log4j.properties"
 
   if [[ "${HADOOP_DAEMON_MODE}" == "default" ]] ||
      [[ "${HADOOP_DAEMON_MODE}" == "start" ]]; then

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java

@@ -224,7 +224,7 @@ boolean deadNodesContain(DatanodeInfo nodeInfo) {
   }
 
   /**
-   * Grab the open-file info from namenode
+   * Grab the open-file info from namenode.
    * @param refreshLocatedBlocks whether to re-fetch locatedblocks
    */
   void openInfo(boolean refreshLocatedBlocks) throws IOException {
@@ -940,7 +940,8 @@ private DNAddrPair chooseDataNode(LocatedBlock block,
    * @return Returns chosen DNAddrPair; Can be null if refetchIfRequired is
    * false.
    */
-  private DNAddrPair chooseDataNode(LocatedBlock block,
+  @VisibleForTesting
+  DNAddrPair chooseDataNode(LocatedBlock block,
       Collection<DatanodeInfo> ignoredNodes, boolean refetchIfRequired)
       throws IOException {
     while (true) {
@@ -955,6 +956,14 @@ private DNAddrPair chooseDataNode(LocatedBlock block,
     }
   }
 
+  /**
+   * RefetchLocations should only be called when there are no active requests
+   * to datanodes. In the hedged read case this means futures should be empty.
+   * @param block The locatedBlock to get new datanode locations for.
+   * @param ignoredNodes A list of ignored nodes. This list can be null and can be cleared.
+   * @return the locatedBlock with updated datanode locations.
+   * @throws IOException
+   */
   private LocatedBlock refetchLocations(LocatedBlock block,
       Collection<DatanodeInfo> ignoredNodes) throws IOException {
     String errMsg = getBestNodeDNAddrPairErrorString(block.getLocations(),
@@ -999,13 +1008,24 @@ private LocatedBlock refetchLocations(LocatedBlock block,
       throw new InterruptedIOException(
           "Interrupted while choosing DataNode for read.");
     }
-    clearLocalDeadNodes(); //2nd option is to remove only nodes[blockId]
+    clearCachedNodeState(ignoredNodes);
     openInfo(true);
     block = refreshLocatedBlock(block);
     failures++;
     return block;
   }
 
+  /**
+   * Clear both the dead nodes and the ignored nodes
+   * @param ignoredNodes is cleared
+   */
+  private void clearCachedNodeState(Collection<DatanodeInfo> ignoredNodes) {
+    clearLocalDeadNodes(); //2nd option is to remove only nodes[blockId]
+    if (ignoredNodes != null) {
+      ignoredNodes.clear();
+    }
+  }
+
   /**
    * Get the best node from which to stream the data.
    * @param block LocatedBlock, containing nodes in priority order.
@@ -1337,8 +1357,12 @@ private void hedgedFetchBlockByteRange(LocatedBlock block, long start,
         } catch (InterruptedException ie) {
           // Ignore and retry
         }
-        if (refetch) {
-          refetchLocations(block, ignored);
+        // If refetch is true, then all nodes are in deadNodes or ignoredNodes.
+        // We should loop through all futures and remove them, so we do not
+        // have concurrent requests to the same node.
+        // Once all futures are cleared, we can clear the ignoredNodes and retry.
+        if (refetch && futures.isEmpty()) {
+          block = refetchLocations(block, ignored);
         }
         // We got here if exception. Ignore this node on next go around IFF
         // we found a chosenNode to hedge read against.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RBFConfigKeys.java

@@ -201,6 +201,10 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
       FEDERATION_ROUTER_PREFIX + "observer.federated.state.propagation.maxsize";
   public static final int DFS_ROUTER_OBSERVER_FEDERATED_STATE_PROPAGATION_MAXSIZE_DEFAULT = 5;
 
+  public static final String DFS_ROUTER_OBSERVER_STATE_ID_REFRESH_PERIOD_KEY =
+      FEDERATION_ROUTER_PREFIX + "observer.state.id.refresh.period";
+  public static final String DFS_ROUTER_OBSERVER_STATE_ID_REFRESH_PERIOD_DEFAULT = "15s";
+
   public static final String FEDERATION_STORE_SERIALIZER_CLASS =
       FEDERATION_STORE_PREFIX + "serializer";
   public static final Class<StateStoreSerializerPBImpl>

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java

@@ -57,6 +57,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAccumulator;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -86,6 +87,7 @@
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.Time;
 import org.eclipse.jetty.util.ajax.JSON;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -136,6 +138,14 @@ public class RouterRpcClient {
   private final boolean observerReadEnabledDefault;
   /** Nameservice specific overrides of the default setting for enabling observer reads. */
   private HashSet<String> observerReadEnabledOverrides = new HashSet<>();
+  /**
+   * Period to refresh namespace stateID using active namenode.
+   * This ensures the namespace stateID is fresh even when an
+   * observer is trailing behind.
+   */
+  private long activeNNStateIdRefreshPeriodMs;
+  /** Last msync times for each namespace. */
+  private final ConcurrentHashMap<String, LongAccumulator> lastActiveNNRefreshTimes;
 
   /** Pattern to parse a stack trace line. */
   private static final Pattern STACK_TRACE_PATTERN =
@@ -211,13 +221,25 @@ public RouterRpcClient(Configuration conf, Router router,
     this.observerReadEnabledDefault = conf.getBoolean(
         RBFConfigKeys.DFS_ROUTER_OBSERVER_READ_DEFAULT_KEY,
         RBFConfigKeys.DFS_ROUTER_OBSERVER_READ_DEFAULT_VALUE);
-    String[] observerReadOverrides = conf.getStrings(RBFConfigKeys.DFS_ROUTER_OBSERVER_READ_OVERRIDES);
+    String[] observerReadOverrides =
+        conf.getStrings(RBFConfigKeys.DFS_ROUTER_OBSERVER_READ_OVERRIDES);
     if (observerReadOverrides != null) {
       observerReadEnabledOverrides.addAll(Arrays.asList(observerReadOverrides));
     }
     if (this.observerReadEnabledDefault) {
       LOG.info("Observer read is enabled for router.");
     }
+    this.activeNNStateIdRefreshPeriodMs = conf.getTimeDuration(
+        RBFConfigKeys.DFS_ROUTER_OBSERVER_STATE_ID_REFRESH_PERIOD_KEY,
+        RBFConfigKeys.DFS_ROUTER_OBSERVER_STATE_ID_REFRESH_PERIOD_DEFAULT,
+        TimeUnit.SECONDS, TimeUnit.MILLISECONDS);
+    if (activeNNStateIdRefreshPeriodMs < 0) {
+      LOG.info("Periodic stateId freshness check is disabled"
+              + " since '{}' is {}ms, which is less than 0.",
+          RBFConfigKeys.DFS_ROUTER_OBSERVER_STATE_ID_REFRESH_PERIOD_KEY,
+          activeNNStateIdRefreshPeriodMs);
+    }
+    this.lastActiveNNRefreshTimes = new ConcurrentHashMap<>();
   }
 
   /**
@@ -1707,10 +1729,13 @@ private List<? extends FederationNamenodeContext> getOrderedNamenodes(String nsI
       boolean isObserverRead) throws IOException {
     final List<? extends FederationNamenodeContext> namenodes;
 
-    if (RouterStateIdContext.getClientStateIdFromCurrentCall(nsId) > Long.MIN_VALUE) {
-      namenodes = namenodeResolver.getNamenodesForNameserviceId(nsId, isObserverRead);
-    } else {
-      namenodes = namenodeResolver.getNamenodesForNameserviceId(nsId, false);
+    boolean listObserverNamenodesFirst = isObserverRead
+        && isNamespaceStateIdFresh(nsId)
+        && (RouterStateIdContext.getClientStateIdFromCurrentCall(nsId) > Long.MIN_VALUE);
+    namenodes = namenodeResolver.getNamenodesForNameserviceId(nsId, listObserverNamenodesFirst);
+    if (!listObserverNamenodesFirst) {
+      // Refresh time of last call to active NameNode.
+      getTimeOfLastCallToActive(nsId).accumulate(Time.monotonicNow());
     }
 
     if (namenodes == null || namenodes.isEmpty()) {
@@ -1721,7 +1746,8 @@ private List<? extends FederationNamenodeContext> getOrderedNamenodes(String nsI
   }
 
   private boolean isObserverReadEligible(String nsId, Method method) {
-    boolean isReadEnabledForNamespace = observerReadEnabledDefault != observerReadEnabledOverrides.contains(nsId);
+    boolean isReadEnabledForNamespace =
+        observerReadEnabledDefault != observerReadEnabledOverrides.contains(nsId);
     return isReadEnabledForNamespace && isReadCall(method);
   }
 
@@ -1735,4 +1761,24 @@ private static boolean isReadCall(Method method) {
     }
     return !method.getAnnotationsByType(ReadOnly.class)[0].activeOnly();
   }
+
+  /**
+   * Checks and sets last refresh time for a namespace's stateId.
+   * Returns true if refresh time is newer than threshold.
+   * Otherwise, return false and call should be handled by active namenode.
+   * @param nsId namespaceID
+   */
+  @VisibleForTesting
+  boolean isNamespaceStateIdFresh(String nsId) {
+    if (activeNNStateIdRefreshPeriodMs < 0) {
+      return true;
+    }
+    long timeSinceRefreshMs = Time.monotonicNow() - getTimeOfLastCallToActive(nsId).get();
+    return (timeSinceRefreshMs <= activeNNStateIdRefreshPeriodMs);
+  }
+
+  private LongAccumulator getTimeOfLastCallToActive(String namespaceId) {
+    return lastActiveNNRefreshTimes
+        .computeIfAbsent(namespaceId, key -> new LongAccumulator(Math::max, 0));
+  }
 }