Merge branch 'apache:trunk' into YARN-11226-V3

Author: slfan1989

LICENSE-binary

@@ -324,7 +324,7 @@ org.apache.htrace:htrace-core:3.1.0-incubating
 org.apache.htrace:htrace-core4:4.1.0-incubating
 org.apache.httpcomponents:httpclient:4.5.6
 org.apache.httpcomponents:httpcore:4.4.10
-org.apache.kafka:kafka-clients:2.8.1
+org.apache.kafka:kafka-clients:2.8.2
 org.apache.kerby:kerb-admin:2.0.2
 org.apache.kerby:kerb-client:2.0.2
 org.apache.kerby:kerb-common:2.0.2

hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/util/PlatformName.java

@@ -18,6 +18,10 @@
 
 package org.apache.hadoop.util;
 
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.Arrays;
+
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 
@@ -33,21 +37,71 @@ public class PlatformName {
    * per the java-vm.
    */
   public static final String PLATFORM_NAME =
-      (System.getProperty("os.name").startsWith("Windows")
-      ? System.getenv("os") : System.getProperty("os.name"))
-      + "-" + System.getProperty("os.arch")
-      + "-" + System.getProperty("sun.arch.data.model");
+      (System.getProperty("os.name").startsWith("Windows") ?
+      System.getenv("os") : System.getProperty("os.name"))
+      + "-" + System.getProperty("os.arch") + "-"
+      + System.getProperty("sun.arch.data.model");
 
   /**
    * The java vendor name used in this platform.
    */
   public static final String JAVA_VENDOR_NAME = System.getProperty("java.vendor");
 
+  /**
+   * Define a system class accessor that is open to changes in underlying implementations
+   * of the system class loader modules.
+   */
+  private static final class SystemClassAccessor extends ClassLoader {
+    public Class<?> getSystemClass(String className) throws ClassNotFoundException {
+      return findSystemClass(className);
+    }
+  }
+
   /**
    * A public static variable to indicate the current java vendor is
-   * IBM java or not.
+   * IBM and the type is Java Technology Edition which provides its
+   * own implementations of many security packages and Cipher suites.
+   * Note that these are not provided in Semeru runtimes:
+   * See https://developer.ibm.com/languages/java/semeru-runtimes for details.
    */
-  public static final boolean IBM_JAVA = JAVA_VENDOR_NAME.contains("IBM");
+  public static final boolean IBM_JAVA = JAVA_VENDOR_NAME.contains("IBM") &&
+      hasIbmTechnologyEditionModules();
+
+  private static boolean hasIbmTechnologyEditionModules() {
+    return Arrays.asList(
+        "com.ibm.security.auth.module.JAASLoginModule",
+        "com.ibm.security.auth.module.Win64LoginModule",
+        "com.ibm.security.auth.module.NTLoginModule",
+        "com.ibm.security.auth.module.AIX64LoginModule",
+        "com.ibm.security.auth.module.LinuxLoginModule",
+        "com.ibm.security.auth.module.Krb5LoginModule"
+    ).stream().anyMatch((module) -> isSystemClassAvailable(module));
+  }
+
+  /**
+   * In rare cases where different behaviour is performed based on the JVM vendor
+   * this method should be used to test for a unique JVM class provided by the
+   * vendor rather than using the vendor method. For example if on JVM provides a
+   * different Kerberos login module testing for that login module being loadable
+   * before configuring to use it is preferable to using the vendor data.
+   *
+   * @param className the name of a class in the JVM to test for
+   * @return true if the class is available, false otherwise.
+   */
+  private static boolean isSystemClassAvailable(String className) {
+    return AccessController.doPrivileged((PrivilegedAction<Boolean>) () -> {
+      try {
+        // Using ClassLoader.findSystemClass() instead of
+        // Class.forName(className, false, null) because Class.forName with a null
+        // ClassLoader only looks at the boot ClassLoader with Java 9 and above
+        // which doesn't look at all the modules available to the findSystemClass.
+        new SystemClassAccessor().getSystemClass(className);
+        return true;
+      } catch (Exception ignored) {
+        return false;
+      }
+    });
+  }
 
   public static void main(String[] args) {
     System.out.println(PLATFORM_NAME);

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

@@ -60,7 +60,6 @@ public AvroFSInput(final FileContext fc, final Path p) throws IOException {
             FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL)
         .withFileStatus(status)
         .build());
-    fc.open(p);
   }
 
   @Override

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

@@ -25,7 +25,7 @@
 import org.apache.hadoop.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import static org.apache.hadoop.util.PlatformName.JAVA_VENDOR_NAME;
+import static org.apache.hadoop.util.PlatformName.IBM_JAVA;
 
 import javax.net.ssl.HostnameVerifier;
 import javax.net.ssl.HttpsURLConnection;
@@ -102,11 +102,11 @@ public enum Mode { CLIENT, SERVER }
       "ssl.server.exclude.cipher.list";
 
   public static final String KEY_MANAGER_SSLCERTIFICATE =
-      JAVA_VENDOR_NAME.contains("IBM") ? "ibmX509" :
+      IBM_JAVA ? "ibmX509" :
           KeyManagerFactory.getDefaultAlgorithm();
 
   public static final String TRUST_MANAGER_SSLCERTIFICATE =
-      JAVA_VENDOR_NAME.contains("IBM") ? "ibmX509" :
+      IBM_JAVA ? "ibmX509" :
           TrustManagerFactory.getDefaultAlgorithm();
 
   public static final String KEYSTORES_FACTORY_CLASS_KEY =

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ApplicationClassLoader.java

@@ -108,7 +108,7 @@ static URL[] constructUrlsFromClasspath(String classpath)
       throws MalformedURLException {
     List<URL> urls = new ArrayList<URL>();
     for (String element : classpath.split(File.pathSeparator)) {
-      if (element.endsWith("/*")) {
+      if (element.endsWith(File.separator + "*")) {
         List<Path> jars = FileUtil.getJarsInDirectory(element);
         if (!jars.isEmpty()) {
           for (Path jar: jars) {

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -2168,9 +2168,8 @@ The switch to turn S3A auditing on or off.
 
 <property>
   <name>fs.azure.enable.readahead</name>
-  <value>false</value>
-  <description>Disable readahead/prefetching in AbfsInputStream.
-    See HADOOP-18521</description>
+  <value>true</value>
+  <description>Enabled readahead/prefetching in AbfsInputStream.</description>
 </property>
 
 <property>

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

@@ -22,7 +22,17 @@ Purpose
 
 This document describes how to install and configure Hadoop clusters ranging from a few nodes to extremely large clusters with thousands of nodes. To play with Hadoop, you may first want to install it on a single machine (see [Single Node Setup](./SingleCluster.html)).
 
-This document does not cover advanced topics such as [Security](./SecureMode.html) or High Availability.
+This document does not cover advanced topics such as High Availability.
+
+*Important*: all production Hadoop clusters use Kerberos to authenticate callers
+and secure access to HDFS data as well as restriction access to computation
+services (YARN etc.). 
+
+These instructions do not cover integration with any Kerberos services,
+-everyone bringing up a production cluster should include connecting to their
+organisation's Kerberos infrastructure as a key part of the deployment.
+
+See [Security](./SecureMode.html) for details on how to secure a cluster. 
 
 Prerequisites
 -------------

hadoop-common-project/hadoop-common/src/site/markdown/SingleCluster.md.vm

@@ -26,15 +26,22 @@ Purpose
 
 This document describes how to set up and configure a single-node Hadoop installation so that you can quickly perform simple operations using Hadoop MapReduce and the Hadoop Distributed File System (HDFS).
 
+
+*Important*: all production Hadoop clusters use Kerberos to authenticate callers
+and secure access to HDFS data as well as restriction access to computation
+services (YARN etc.). 
+
+These instructions do not cover integration with any Kerberos services,
+-everyone bringing up a production cluster should include connecting to their
+organisation's Kerberos infrastructure as a key part of the deployment.
+
 Prerequisites
 -------------
 
 $H3 Supported Platforms
 
 * GNU/Linux is supported as a development and production platform. Hadoop has been demonstrated on GNU/Linux clusters with 2000 nodes.
 
-* Windows is also a supported platform but the followings steps are for Linux only. To set up Hadoop on Windows, see [wiki page](http://wiki.apache.org/hadoop/Hadoop2OnWindows).
-
 $H3 Required Software
 
 Required software for Linux include:

hadoop-common-project/hadoop-minikdc/src/test/java/org/apache/hadoop/minikdc/TestMiniKdc.java

@@ -38,8 +38,35 @@
 import java.util.Arrays;
 
 public class TestMiniKdc extends KerberosSecurityTestcase {
-  private static final boolean IBM_JAVA = System.getProperty("java.vendor")
-      .contains("IBM");
+  private static final boolean IBM_JAVA = shouldUseIbmPackages();
+  // duplicated to avoid cycles in the build
+  private static boolean shouldUseIbmPackages() {
+    final List<String> ibmTechnologyEditionSecurityModules = Arrays.asList(
+        "com.ibm.security.auth.module.JAASLoginModule",
+        "com.ibm.security.auth.module.Win64LoginModule",
+        "com.ibm.security.auth.module.NTLoginModule",
+        "com.ibm.security.auth.module.AIX64LoginModule",
+        "com.ibm.security.auth.module.LinuxLoginModule",
+        "com.ibm.security.auth.module.Krb5LoginModule"
+    );
+
+    if (System.getProperty("java.vendor").contains("IBM")) {
+      return ibmTechnologyEditionSecurityModules
+          .stream().anyMatch((module) -> isSystemClassAvailable(module));
+    }
+
+    return false;
+  }
+
+  private static boolean isSystemClassAvailable(String className) {
+    try {
+      Class.forName(className);
+      return true;
+    } catch (Exception ignored) {
+      return false;
+    }
+  }
+
   @Test
   public void testMiniKdcStart() {
     MiniKdc kdc = getKdc();
@@ -117,9 +144,9 @@ public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
       options.put("debug", "true");
 
       return new AppConfigurationEntry[]{
-              new AppConfigurationEntry(getKrb5LoginModuleName(),
-                      AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
-                      options)};
+          new AppConfigurationEntry(getKrb5LoginModuleName(),
+                AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+                options)};
     }
   }
 

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

@@ -20,13 +20,19 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.RouterFederatedStateProto;
 import org.apache.hadoop.ipc.AlignmentContext;
 import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcRequestHeaderProto;
 import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.atomic.LongAccumulator;
 import org.apache.hadoop.thirdparty.protobuf.ByteString;
+import org.apache.hadoop.thirdparty.protobuf.InvalidProtocolBufferException;
 
 /**
  * Global State Id context for the client.
@@ -77,12 +83,46 @@ public void updateResponseState(RpcResponseHeaderProto.Builder header) {
   @Override
   public synchronized void receiveResponseState(RpcResponseHeaderProto header) {
     if (header.hasRouterFederatedState()) {
-      routerFederatedState = header.getRouterFederatedState();
+      routerFederatedState = mergeRouterFederatedState(
+          this.routerFederatedState, header.getRouterFederatedState());
     } else {
       lastSeenStateId.accumulate(header.getStateId());
     }
   }
 
+  /**
+   * Utility function to parse routerFederatedState field in RPC headers.
+   */
+  public static Map<String, Long> getRouterFederatedStateMap(ByteString byteString) {
+    if (byteString != null) {
+      try {
+        RouterFederatedStateProto federatedState = RouterFederatedStateProto.parseFrom(byteString);
+        return federatedState.getNamespaceStateIdsMap();
+      } catch (InvalidProtocolBufferException e) {
+        // Ignore this exception and will return an empty map
+      }
+    }
+    return Collections.emptyMap();
+  }
+
+  /**
+   * Merge state1 and state2 to get the max value for each namespace.
+   * @param state1 input ByteString.
+   * @param state2 input ByteString.
+   * @return one ByteString object which contains the max value of each namespace.
+   */
+  public static ByteString mergeRouterFederatedState(ByteString state1, ByteString state2) {
+    Map<String, Long> mapping1 = new HashMap<>(getRouterFederatedStateMap(state1));
+    Map<String, Long> mapping2 = getRouterFederatedStateMap(state2);
+    mapping2.forEach((k, v) -> {
+      long localValue = mapping1.getOrDefault(k, 0L);
+      mapping1.put(k, Math.max(v, localValue));
+    });
+    RouterFederatedStateProto.Builder federatedBuilder = RouterFederatedStateProto.newBuilder();
+    mapping1.forEach(federatedBuilder::putNamespaceStateIds);
+    return federatedBuilder.build().toByteString();
+  }
+
   /**
    * Client side implementation for providing state alignment info in requests.
    */
@@ -106,4 +146,9 @@ public long receiveRequestState(RpcRequestHeaderProto header, long threshold)
     // Do nothing.
     return 0;
   }
+
+  @VisibleForTesting
+  public ByteString getRouterFederatedState() {
+    return this.routerFederatedState;
+  }
 }