Merge branch 'apache:trunk' into YARN-11313

Author: slfan1989

LICENSE-binary

@@ -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
 
 

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/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/statistics/IOStatisticsContext.java

@@ -20,6 +20,8 @@
 
 import org.apache.hadoop.fs.statistics.impl.IOStatisticsContextIntegration;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * An interface defined to capture thread-level IOStatistics by using per
  * thread context.
@@ -67,7 +69,11 @@ public interface IOStatisticsContext extends IOStatisticsSource {
    * @return instance of IOStatisticsContext for the context.
    */
   static IOStatisticsContext getCurrentIOStatisticsContext() {
-    return IOStatisticsContextIntegration.getCurrentIOStatisticsContext();
+    // the null check is just a safety check to highlight exactly where a null value would
+    // be returned if HADOOP-18456 has resurfaced.
+    return requireNonNull(
+        IOStatisticsContextIntegration.getCurrentIOStatisticsContext(),
+        "Null IOStatisticsContext");
   }
 
   /**

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

@@ -100,7 +100,10 @@ private IOStatisticsContextIntegration() {}
    * @return an instance of IOStatisticsContext.
    */
   private static IOStatisticsContext createNewInstance(Long key) {
-    return new IOStatisticsContextImpl(key, INSTANCE_ID.getAndIncrement());
+    IOStatisticsContextImpl instance =
+        new IOStatisticsContextImpl(key, INSTANCE_ID.getAndIncrement());
+    LOG.debug("Created instance {}", instance);
+    return instance;
   }
 
   /**
@@ -131,9 +134,11 @@ public static void setThreadIOStatisticsContext(
       IOStatisticsContext statisticsContext) {
     if (isThreadIOStatsEnabled) {
       if (statisticsContext == null) {
+        // new value is null, so remove it
         ACTIVE_IOSTATS_CONTEXT.removeForCurrentThread();
-      }
-      if (ACTIVE_IOSTATS_CONTEXT.getForCurrentThread() != statisticsContext) {
+      } else {
+        // the setter is efficient in that it does not create a new
+        // reference if the context is unchanged.
         ACTIVE_IOSTATS_CONTEXT.setForCurrentThread(statisticsContext);
       }
     }

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

@@ -124,12 +124,28 @@ public static void setCallIdAndRetryCount(int cid, int rc,
     Preconditions.checkArgument(cid != RpcConstants.INVALID_CALL_ID);
     Preconditions.checkState(callId.get() == null);
     Preconditions.checkArgument(rc != RpcConstants.INVALID_RETRY_COUNT);
+    setCallIdAndRetryCountUnprotected(cid, rc, externalHandler);
+  }
 
+  public static void setCallIdAndRetryCountUnprotected(Integer cid, int rc,
+      Object externalHandler) {
     callId.set(cid);
     retryCount.set(rc);
     EXTERNAL_CALL_HANDLER.set(externalHandler);
   }
 
+  public static int getCallId() {
+    return callId.get() != null ? callId.get() : nextCallId();
+  }
+
+  public static int getRetryCount() {
+    return retryCount.get() != null ? retryCount.get() : 0;
+  }
+
+  public static Object getExternalHandler() {
+    return EXTERNAL_CALL_HANDLER.get();
+  }
+
   private final ConcurrentMap<ConnectionId, Connection> connections =
       new ConcurrentHashMap<>();
   private final Object putLock = new Object();