YARN-11349. [Federation] Router Support DelegationToken With SQL.

LICENSE-binary

@@ -251,7 +251,7 @@ commons-collections:commons-collections:3.2.2
 commons-daemon:commons-daemon:1.0.13
 commons-io:commons-io:2.8.0
 commons-logging:commons-logging:1.1.3
-commons-net:commons-net:3.8.0
+commons-net:commons-net:3.9.0
 de.ruedigermoeller:fst:2.50
 io.grpc:grpc-api:1.26.0
 io.grpc:grpc-context:1.26.0
@@ -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/fs/VectoredReadUtils.java

@@ -307,9 +307,16 @@ public static ByteBuffer sliceTo(ByteBuffer readData, long readOffset,
                                    FileRange request) {
     int offsetChange = (int) (request.getOffset() - readOffset);
     int requestLength = request.getLength();
+    // Create a new buffer that is backed by the original contents
+    // The buffer will have position 0 and the same limit as the original one
     readData = readData.slice();
+    // Change the offset and the limit of the buffer as the reader wants to see
+    // only relevant data
     readData.position(offsetChange);
     readData.limit(offsetChange + requestLength);
+    // Create a new buffer after the limit change so that only that portion of the data is
+    // returned to the reader.
+    readData = readData.slice();
     return readData;
   }
 

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

@@ -90,6 +90,11 @@ private AuditConstants() {
    */
   public static final String PARAM_PROCESS = "ps";
 
+  /**
+   * Header: Range for GET request data: {@value}.
+   */
+  public static final String PARAM_RANGE = "rg";
+
   /**
    * Task Attempt ID query header: {@value}.
    */

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

@@ -0,0 +1,97 @@
+/*
+ * 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.fs.impl;
+
+import java.lang.ref.WeakReference;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.metrics2.MetricsCollector;
+import org.apache.hadoop.metrics2.MetricsSource;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A weak referenced metrics source which avoids hanging on to large objects
+ * if somehow they don't get fully closed/cleaned up.
+ * The JVM may clean up all objects which are only weakly referenced whenever
+ * it does a GC, <i>even if there is no memory pressure</i>.
+ * To avoid these refs being removed, always keep a strong reference around
+ * somewhere.
+ */
+@InterfaceAudience.Private
+public class WeakRefMetricsSource implements MetricsSource {
+
+  /**
+   * Name to know when unregistering.
+   */
+  private final String name;
+
+  /**
+   * Underlying metrics source.
+   */
+  private final WeakReference<MetricsSource> sourceWeakReference;
+
+  /**
+   * Constructor.
+   * @param name Name to know when unregistering.
+   * @param source metrics source
+   */
+  public WeakRefMetricsSource(final String name, final MetricsSource source) {
+    this.name = name;
+    this.sourceWeakReference = new WeakReference<>(requireNonNull(source));
+  }
+
+  /**
+   * If the weak reference is non null, update the metrics.
+   * @param collector to contain the resulting metrics snapshot
+   * @param all if true, return all metrics even if unchanged.
+   */
+  @Override
+  public void getMetrics(final MetricsCollector collector, final boolean all) {
+    MetricsSource metricsSource = sourceWeakReference.get();
+    if (metricsSource != null) {
+      metricsSource.getMetrics(collector, all);
+    }
+  }
+
+  /**
+   * Name to know when unregistering.
+   * @return the name passed in during construction.
+   */
+  public String getName() {
+    return name;
+  }
+
+  /**
+   * Get the source, will be null if the reference has been GC'd
+   * @return the source reference
+   */
+  public MetricsSource getSource() {
+    return sourceWeakReference.get();
+  }
+
+  @Override
+  public String toString() {
+    return "WeakRefMetricsSource{" +
+        "name='" + name + '\'' +
+        ", sourceWeakReference is " +
+        (sourceWeakReference.get() == null ? "unset" : "set") +
+        '}';
+  }
+}

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

@@ -190,7 +190,7 @@ public long incrementCounter(final String key, final long value) {
       return counter.get();
     } else {
       long l = incAtomicLong(counter, value);
-      LOG.debug("Incrementing counter {} by {} with final value {}",
+      LOG.trace("Incrementing counter {} by {} with final value {}",
           key, value, l);
       return l;
     }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/log/LogThrottlingHelper.java

@@ -262,9 +262,15 @@ public LogAction record(String recorderName, long currentTimeMs,
     if (primaryRecorderName.equals(recorderName) &&
         currentTimeMs - minLogPeriodMs >= lastLogTimestampMs) {
       lastLogTimestampMs = currentTimeMs;
-      for (LoggingAction log : currentLogs.values()) {
-        log.setShouldLog();
-      }
+      currentLogs.replaceAll((key, log) -> {
+        LoggingAction newLog = log;
+        if (log.hasLogged()) {
+          // create a fresh log since the old one has already been logged
+          newLog = new LoggingAction(log.getValueCount());
+        }
+        newLog.setShouldLog();
+        return newLog;
+      });
     }
     if (currentLog.shouldLog()) {
       currentLog.setHasLogged();
@@ -357,6 +363,10 @@ private void setHasLogged() {
       hasLogged = true;
     }
 
+    private int getValueCount() {
+      return stats.length;
+    }
+
     private void recordValues(double... values) {
       if (values.length != stats.length) {
         throw new IllegalArgumentException("received " + values.length +

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

@@ -38,6 +38,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.util.Shell.bashQuote;
+
 /**
  * A simple shell-based implementation of {@link IdMappingServiceProvider} 
  * Map id to user name or group name. It does update every 15 minutes. Only a
@@ -472,26 +474,27 @@ synchronized private void updateMapIncr(final String name,
 
     boolean updated = false;
     updateStaticMapping();
+    String name2 = bashQuote(name);
 
     if (OS.startsWith("Linux") || OS.equals("SunOS") || OS.contains("BSD")) {
       if (isGrp) {
         updated = updateMapInternal(gidNameMap, "group",
-            getName2IdCmdNIX(name, true), ":",
+            getName2IdCmdNIX(name2, true), ":",
             staticMapping.gidMapping);
       } else {
         updated = updateMapInternal(uidNameMap, "user",
-            getName2IdCmdNIX(name, false), ":",
+            getName2IdCmdNIX(name2, false), ":",
             staticMapping.uidMapping);
       }
     } else {
       // Mac
       if (isGrp) {        
         updated = updateMapInternal(gidNameMap, "group",
-            getName2IdCmdMac(name, true), "\\s+",
+            getName2IdCmdMac(name2, true), "\\s+",
             staticMapping.gidMapping);
       } else {
         updated = updateMapInternal(uidNameMap, "user",
-            getName2IdCmdMac(name, false), "\\s+",
+            getName2IdCmdMac(name2, false), "\\s+",
             staticMapping.uidMapping);
       }
     }

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/security/token/delegation/AbstractDelegationTokenSecretManager.java

@@ -866,9 +866,9 @@ private String getTokenRealOwner(TokenIdent id) {
   /**
    * Add token stats to the owner to token count mapping.
    *
-   * @param id
+   * @param id token id.
    */
-  private void addTokenForOwnerStats(TokenIdent id) {
+  protected void addTokenForOwnerStats(TokenIdent id) {
     String realOwner = getTokenRealOwner(id);
     tokenOwnerStats.put(realOwner,
         tokenOwnerStats.getOrDefault(realOwner, 0L)+1);

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/java/org/apache/hadoop/util/Shell.java

@@ -146,7 +146,8 @@ public static void checkWindowsCommandLineLength(String...commands)
    * @param arg the argument to quote
    * @return the quoted string
    */
-  static String bashQuote(String arg) {
+  @InterfaceAudience.Private
+  public static String bashQuote(String arg) {
     StringBuilder buffer = new StringBuilder(arg.length() + 2);
     buffer.append('\'')
         .append(arg.replace("'", "'\\''"))