Merge branch 'apache:trunk' into YARN-11247

Author: slfan1989

LICENSE-binary

@@ -305,12 +305,12 @@ net.minidev:json-smart:2.4.7
 org.apache.avro:avro:1.9.2
 org.apache.commons:commons-collections4:4.2
 org.apache.commons:commons-compress:1.21
-org.apache.commons:commons-configuration2:2.1.1
+org.apache.commons:commons-configuration2:2.8.0
 org.apache.commons:commons-csv:1.0
 org.apache.commons:commons-digester:1.8.1
 org.apache.commons:commons-lang3:3.12.0
 org.apache.commons:commons-math3:3.6.1
-org.apache.commons:commons-text:1.4
+org.apache.commons:commons-text:1.9
 org.apache.commons:commons-validator:1.6
 org.apache.curator:curator-client:5.2.0
 org.apache.curator:curator-framework:5.2.0
@@ -362,7 +362,7 @@ org.ehcache:ehcache:3.3.1
 org.lz4:lz4-java:1.7.1
 org.objenesis:objenesis:2.6
 org.xerial.snappy:snappy-java:1.0.5
-org.yaml:snakeyaml:1.31:
+org.yaml:snakeyaml:1.32
 org.wildfly.openssl:wildfly-openssl:1.0.7.Final
 
 
@@ -523,7 +523,7 @@ junit:junit:4.13.2
 HSQL License
 ------------
 
-org.hsqldb:hsqldb:2.3.4
+org.hsqldb:hsqldb:2.5.2
 
 
 JDOM License

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/ConfServlet.java

@@ -98,7 +98,7 @@ static void writeResponse(Configuration conf,
     if (FORMAT_JSON.equals(format)) {
       Configuration.dumpConfiguration(conf, propertyName, out);
     } else if (FORMAT_XML.equals(format)) {
-      conf.writeXml(propertyName, out);
+      conf.writeXml(propertyName, out, conf);
     } else {
       throw new BadFormatException("Bad format: " + format);
     }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/ConfigRedactor.java

@@ -37,6 +37,7 @@
 public class ConfigRedactor {
 
   private static final String REDACTED_TEXT = "<redacted>";
+  private static final String REDACTED_XML = "******";
 
   private List<Pattern> compiledPatterns;
 
@@ -84,4 +85,19 @@ private boolean configIsSensitive(String key) {
     }
     return false;
   }
+
+  /**
+   * Given a key / value pair, decides whether or not to redact and returns
+   * either the original value or text indicating it has been redacted.
+   *
+   * @param key param key.
+   * @param value param value, will return if conditions permit.
+   * @return Original value, or text indicating it has been redacted
+   */
+  public String redactXml(String key, String value) {
+    if (configIsSensitive(key)) {
+      return REDACTED_XML;
+    }
+    return value;
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java

@@ -3593,11 +3593,13 @@ public void writeXml(Writer out) throws IOException {
    * </ul>
    * @param propertyName xml property name.
    * @param out the writer to write to.
+   * @param config configuration.
    * @throws IOException raised on errors performing I/O.
    */
-  public void writeXml(@Nullable String propertyName, Writer out)
+  public void writeXml(@Nullable String propertyName, Writer out, Configuration config)
       throws IOException, IllegalArgumentException {
-    Document doc = asXmlDocument(propertyName);
+    ConfigRedactor redactor = config != null ? new ConfigRedactor(this) : null;
+    Document doc = asXmlDocument(propertyName, redactor);
 
     try {
       DOMSource source = new DOMSource(doc);
@@ -3614,11 +3616,16 @@ public void writeXml(@Nullable String propertyName, Writer out)
     }
   }
 
+  public void writeXml(@Nullable String propertyName, Writer out)
+      throws IOException, IllegalArgumentException {
+    writeXml(propertyName, out, null);
+  }
+
   /**
    * Return the XML DOM corresponding to this Configuration.
    */
-  private synchronized Document asXmlDocument(@Nullable String propertyName)
-      throws IOException, IllegalArgumentException {
+  private synchronized Document asXmlDocument(@Nullable String propertyName,
+      ConfigRedactor redactor) throws IOException, IllegalArgumentException {
     Document doc;
     try {
       doc = DocumentBuilderFactory
@@ -3641,13 +3648,13 @@ private synchronized Document asXmlDocument(@Nullable String propertyName)
             propertyName + " not found");
       } else {
         // given property is found, write single property
-        appendXMLProperty(doc, conf, propertyName);
+        appendXMLProperty(doc, conf, propertyName, redactor);
         conf.appendChild(doc.createTextNode("\n"));
       }
     } else {
       // append all elements
       for (Enumeration<Object> e = properties.keys(); e.hasMoreElements();) {
-        appendXMLProperty(doc, conf, (String)e.nextElement());
+        appendXMLProperty(doc, conf, (String)e.nextElement(), redactor);
         conf.appendChild(doc.createTextNode("\n"));
       }
     }
@@ -3663,7 +3670,7 @@ private synchronized Document asXmlDocument(@Nullable String propertyName)
    * @param propertyName
    */
   private synchronized void appendXMLProperty(Document doc, Element conf,
-      String propertyName) {
+      String propertyName, ConfigRedactor redactor) {
     // skip writing if given property name is empty or null
     if (!Strings.isNullOrEmpty(propertyName)) {
       String value = properties.getProperty(propertyName);
@@ -3676,8 +3683,11 @@ private synchronized void appendXMLProperty(Document doc, Element conf,
         propNode.appendChild(nameNode);
 
         Element valueNode = doc.createElement("value");
-        valueNode.appendChild(doc.createTextNode(
-            properties.getProperty(propertyName)));
+        String propertyValue = properties.getProperty(propertyName);
+        if (redactor != null) {
+          propertyValue = redactor.redactXml(propertyName, propertyValue);
+        }
+        valueNode.appendChild(doc.createTextNode(propertyValue));
         propNode.appendChild(valueNode);
 
         Element finalNode = doc.createElement("final");

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

@@ -1000,6 +1000,7 @@ public class CommonConfigurationKeysPublic {
       String.join(",",
           "secret$",
           "password$",
+          "username$",
           "ssl.keystore.pass$",
           "fs.s3.*[Ss]ecret.?[Kk]ey",
           "fs.s3a.*.server-side-encryption.key",

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

@@ -396,6 +396,10 @@ public Path getLocalPathForWrite(String pathStr, long size,
       Context ctx = confChanged(conf);
       int numDirs = ctx.localDirs.length;
       int numDirsSearched = 0;
+      // Max capacity in any directory
+      long maxCapacity = 0;
+      String errorText = null;
+      IOException diskException = null;
       //remove the leading slash from the path (to make sure that the uri
       //resolution results in a valid path on the dir being checked)
       if (pathStr.startsWith("/")) {
@@ -444,9 +448,18 @@ public Path getLocalPathForWrite(String pathStr, long size,
         int dirNum = ctx.getAndIncrDirNumLastAccessed(randomInc);
         while (numDirsSearched < numDirs) {
           long capacity = ctx.dirDF[dirNum].getAvailable();
+          if (capacity > maxCapacity) {
+            maxCapacity = capacity;
+          }
           if (capacity > size) {
-            returnPath =
-                createPath(ctx.localDirs[dirNum], pathStr, checkWrite);
+            try {
+              returnPath = createPath(ctx.localDirs[dirNum], pathStr,
+                  checkWrite);
+            } catch (IOException e) {
+              errorText = e.getMessage();
+              diskException = e;
+              LOG.debug("DiskException caught for dir {}", ctx.localDirs[dirNum], e);
+            }
             if (returnPath != null) {
               ctx.getAndIncrDirNumLastAccessed(numDirsSearched);
               break;
@@ -462,8 +475,13 @@ public Path getLocalPathForWrite(String pathStr, long size,
       }
 
       //no path found
-      throw new DiskErrorException("Could not find any valid local " +
-          "directory for " + pathStr);
+      String newErrorText = "Could not find any valid local directory for " +
+          pathStr + " with requested size " + size +
+          " as the max capacity in any directory is " + maxCapacity;
+      if (errorText != null) {
+        newErrorText = newErrorText + " due to " + errorText;
+      }
+      throw new DiskErrorException(newErrorText, diskException);
     }
 
     /** Creates a file on the local FS. Pass size as 

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

@@ -84,7 +84,7 @@ public interface StreamCapabilities {
    * Support for vectored IO api.
    * See {@code PositionedReadable#readVectored(List, IntFunction)}.
    */
-  String VECTOREDIO = "readvectored";
+  String VECTOREDIO = "in:readvectored";
 
   /**
    * Stream abort() capability implemented by {@link Abortable#abort()}.

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

@@ -23,8 +23,10 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.viewfs.ViewFileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import static org.apache.hadoop.fs.viewfs.Constants.*;
 
 /** 
  * Provides a trash facility which supports pluggable Trash policies. 
@@ -94,6 +96,27 @@ public static boolean moveToAppropriateTrash(FileSystem fs, Path p,
       LOG.warn("Failed to get server trash configuration", e);
       throw new IOException("Failed to get server trash configuration", e);
     }
+
+    /*
+     * In HADOOP-18144, we changed getTrashRoot() in ViewFileSystem to return a
+     * viewFS path, instead of a targetFS path. moveToTrash works for
+     * ViewFileSystem now. ViewFileSystem will do path resolution internally by
+     * itself.
+     *
+     * When localized trash flag is enabled:
+     *    1). if fs is a ViewFileSystem, we can initialize Trash() with a
+     *        ViewFileSystem object;
+     *    2). When fs is not a ViewFileSystem, the only place we would need to
+     *        resolve a path is for symbolic links. However, symlink is not
+     *        enabled in Hadoop due to the complexity to support it
+     *        (HADOOP-10019).
+     */
+    if (conf.getBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT,
+        CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT)) {
+      Trash trash = new Trash(fs, conf);
+      return trash.moveToTrash(p);
+    }
+
     Trash trash = new Trash(fullyResolvedFs, conf);
     return trash.moveToTrash(fullyResolvedPath);
   }

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

@@ -25,6 +25,8 @@
 
 import org.apache.hadoop.util.WeakReferenceMap;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * A WeakReferenceMap for threads.
  * @param <V> value type of the map
@@ -36,30 +38,55 @@ public WeakReferenceThreadMap(final Function<? super Long, ? extends V> factory,
     super(factory, referenceLost);
   }
 
+  /**
+   * Get the value for the current thread, creating if needed.
+   * @return an instance.
+   */
   public V getForCurrentThread() {
     return get(currentThreadId());
   }
 
+  /**
+   * Remove the reference for the current thread.
+   * @return any reference value which existed.
+   */
   public V removeForCurrentThread() {
     return remove(currentThreadId());
   }
 
+  /**
+   * Get the current thread ID.
+   * @return thread ID.
+   */
   public long currentThreadId() {
     return Thread.currentThread().getId();
   }
 
+  /**
+   * Set the new value for the current thread.
+   * @param newVal new reference to set for the active thread.
+   * @return the previously set value, possibly null
+   */
   public V setForCurrentThread(V newVal) {
+    requireNonNull(newVal);
     long id = currentThreadId();
 
     // if the same object is already in the map, just return it.
-    WeakReference<V> ref = lookup(id);
-    // Reference value could be set to null. Thus, ref.get() could return
-    // null. Should be handled accordingly while using the returned value.
-    if (ref != null && ref.get() == newVal) {
-      return ref.get();
+    WeakReference<V> existingWeakRef = lookup(id);
+
+    // The looked up reference could be one of
+    // 1. null: nothing there
+    // 2. valid but get() == null : reference lost by GC.
+    // 3. different from the new value
+    // 4. the same as the old value
+    if (resolve(existingWeakRef) == newVal) {
+      // case 4: do nothing, return the new value
+      return newVal;
+    } else {
+      // cases 1, 2, 3: update the map and return the old value
+      return put(id, newVal);
     }
 
-    return put(id, newVal);
   }
 
 }

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

@@ -22,8 +22,13 @@
 import java.util.Locale;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import org.slf4j.Logger;
+
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
+
 /**
  * A FuturePool implementation backed by a java.util.concurrent.ExecutorService.
  *
@@ -37,7 +42,8 @@
  *
  */
 public class ExecutorServiceFuturePool {
-  private ExecutorService executor;
+
+  private final ExecutorService executor;
 
   public ExecutorServiceFuturePool(ExecutorService executor) {
     this.executor = executor;
@@ -64,6 +70,18 @@ public Future<Void> executeRunnable(final Runnable r) {
     return (Future<Void>) executor.submit(r::run);
   }
 
+  /**
+   * Utility to shutdown the {@link ExecutorService} used by this class. Will wait up to a
+   * certain timeout for the ExecutorService to gracefully shutdown.
+   *
+   * @param logger Logger
+   * @param timeout the maximum time to wait
+   * @param unit the time unit of the timeout argument
+   */
+  public void shutdown(Logger logger, long timeout, TimeUnit unit) {
+    HadoopExecutors.shutdown(executor, logger, timeout, unit);
+  }
+
   public String toString() {
     return String.format(Locale.ROOT, "ExecutorServiceFuturePool(executor=%s)", executor);
   }