Merge branch 'apache:trunk' into YARN-11332

Author: slfan1989

LICENSE-binary

@@ -220,7 +220,7 @@ com.cedarsoftware:java-util:1.9.0
 com.cedarsoftware:json-io:2.5.1
 com.fasterxml.jackson.core:jackson-annotations:2.12.7
 com.fasterxml.jackson.core:jackson-core:2.12.7
-com.fasterxml.jackson.core:jackson-databind:2.12.7
+com.fasterxml.jackson.core:jackson-databind:2.12.7.1
 com.fasterxml.jackson.jaxrs:jackson-jaxrs-base:2.12.7
 com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.12.7
 com.fasterxml.jackson.module:jackson-module-jaxb-annotations:2.12.7
@@ -306,7 +306,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.8.0
-org.apache.commons:commons-csv:1.0
+org.apache.commons:commons-csv:1.9.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

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

@@ -44,12 +44,7 @@ public class InstrumentedReadLock extends InstrumentedLock {
    * there can be multiple threads that hold the read lock concurrently.
    */
   private final ThreadLocal<Long> readLockHeldTimeStamp =
-      new ThreadLocal<Long>() {
-    @Override
-    protected Long initialValue() {
-      return Long.MAX_VALUE;
-    };
-  };
+      ThreadLocal.withInitial(() -> Long.MAX_VALUE);
 
   public InstrumentedReadLock(String name, Logger logger,
       ReentrantReadWriteLock readWriteLock,

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

@@ -37,6 +37,9 @@
 @InterfaceStability.Unstable
 public class InstrumentedWriteLock extends InstrumentedLock {
 
+  private final ReentrantReadWriteLock readWriteLock;
+  private volatile long writeLockHeldTimeStamp = 0;
+
   public InstrumentedWriteLock(String name, Logger logger,
       ReentrantReadWriteLock readWriteLock,
       long minLoggingGapMs, long lockWarningThresholdMs) {
@@ -50,5 +53,28 @@ public InstrumentedWriteLock(String name, Logger logger,
       long minLoggingGapMs, long lockWarningThresholdMs, Timer clock) {
     super(name, logger, readWriteLock.writeLock(), minLoggingGapMs,
         lockWarningThresholdMs, clock);
+    this.readWriteLock = readWriteLock;
+  }
+
+  @Override
+  public void unlock() {
+    boolean needReport = readWriteLock.getWriteHoldCount() == 1;
+    long localWriteReleaseTime = getTimer().monotonicNow();
+    long localWriteAcquireTime = writeLockHeldTimeStamp;
+    getLock().unlock();
+    if (needReport) {
+      writeLockHeldTimeStamp = 0;
+      check(localWriteAcquireTime, localWriteReleaseTime, true);
+    }
+  }
+
+  /**
+   * Starts timing for the instrumented write lock.
+   */
+  @Override
+  protected void startLockTiming() {
+    if (readWriteLock.getWriteHoldCount() == 1) {
+      writeLockHeldTimeStamp = getTimer().monotonicNow();
+    }
   }
 }

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

@@ -41,17 +41,18 @@
 @InterfaceStability.Unstable
 public class XMLUtils {
 
-  private static final String DISALLOW_DOCTYPE_DECL =
+  public static final String DISALLOW_DOCTYPE_DECL =
       "http://apache.org/xml/features/disallow-doctype-decl";
-  private static final String LOAD_EXTERNAL_DECL =
+  public static final String LOAD_EXTERNAL_DECL =
       "http://apache.org/xml/features/nonvalidating/load-external-dtd";
-  private static final String EXTERNAL_GENERAL_ENTITIES =
+  public static final String EXTERNAL_GENERAL_ENTITIES =
       "http://xml.org/sax/features/external-general-entities";
-  private static final String EXTERNAL_PARAMETER_ENTITIES =
+  public static final String EXTERNAL_PARAMETER_ENTITIES =
       "http://xml.org/sax/features/external-parameter-entities";
-  private static final String CREATE_ENTITY_REF_NODES =
+  public static final String CREATE_ENTITY_REF_NODES =
       "http://apache.org/xml/features/dom/create-entity-ref-nodes";
-
+  public static final String VALIDATION =
+      "http://xml.org/sax/features/validation";
 
   /**
    * Transform input xml given a stylesheet.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestInstrumentedReadWriteLock.java

@@ -233,4 +233,111 @@ protected void logWarning(long lockHeldTime, SuppressedSnapshot stats) {
     assertEquals(2, wlogged.get());
     assertEquals(1, wsuppresed.get());
   }
+
+
+  /**
+   * Tests the warning when the write lock is held longer than threshold.
+   */
+  @Test(timeout=10000)
+  public void testWriteLockLongHoldingReportWithReentrant() {
+    String testname = name.getMethodName();
+    final AtomicLong time = new AtomicLong(0);
+    Timer mclock = new Timer() {
+      @Override
+      public long monotonicNow() {
+        return time.get();
+      }
+    };
+
+    final AtomicLong wlogged = new AtomicLong(0);
+    final AtomicLong wsuppresed = new AtomicLong(0);
+    final AtomicLong totalHeldTime = new AtomicLong(0);
+    ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(true);
+    InstrumentedWriteLock writeLock = new InstrumentedWriteLock(testname, LOG,
+        readWriteLock, 2000, 300, mclock) {
+      @Override
+      protected void logWarning(long lockHeldTime, SuppressedSnapshot stats) {
+        totalHeldTime.addAndGet(lockHeldTime);
+        wlogged.incrementAndGet();
+        wsuppresed.set(stats.getSuppressedCount());
+      }
+    };
+
+    InstrumentedReadLock readLock = new InstrumentedReadLock(testname, LOG,
+        readWriteLock, 2000, 300, mclock) {
+      @Override
+      protected void logWarning(long lockHeldTime, SuppressedSnapshot stats) {
+        totalHeldTime.addAndGet(lockHeldTime);
+        wlogged.incrementAndGet();
+        wsuppresed.set(stats.getSuppressedCount());
+      }
+    };
+
+    writeLock.lock();   // t = 0
+    time.set(100);
+
+    writeLock.lock();   // t = 100
+    time.set(500);
+
+    writeLock.lock();   // t = 500
+    time.set(2900);
+    writeLock.unlock(); // t = 2900
+
+    readLock.lock();    // t = 2900
+    time.set(3000);
+    readLock.unlock();  // t = 3000
+
+    writeLock.unlock(); // t = 3000
+
+    writeLock.unlock(); // t = 3000
+    assertEquals(1, wlogged.get());
+    assertEquals(0, wsuppresed.get());
+    assertEquals(3000, totalHeldTime.get());
+  }
+
+  /**
+   * Tests the warning when the read lock is held longer than threshold.
+   */
+  @Test(timeout=10000)
+  public void testReadLockLongHoldingReportWithReentrant() {
+    String testname = name.getMethodName();
+    final AtomicLong time = new AtomicLong(0);
+    Timer mclock = new Timer() {
+      @Override
+      public long monotonicNow() {
+        return time.get();
+      }
+    };
+
+    final AtomicLong wlogged = new AtomicLong(0);
+    final AtomicLong wsuppresed = new AtomicLong(0);
+    final AtomicLong totalHelpTime = new AtomicLong(0);
+    ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(true);
+    InstrumentedReadLock readLock = new InstrumentedReadLock(testname, LOG,
+        readWriteLock, 2000, 300, mclock) {
+      @Override
+      protected void logWarning(long lockHeldTime, SuppressedSnapshot stats) {
+        totalHelpTime.addAndGet(lockHeldTime);
+        wlogged.incrementAndGet();
+        wsuppresed.set(stats.getSuppressedCount());
+      }
+    };
+
+    readLock.lock();   // t = 0
+    time.set(100);
+
+    readLock.lock();   // t = 100
+    time.set(500);
+
+    readLock.lock();   // t = 500
+    time.set(3000);
+    readLock.unlock(); // t = 3000
+
+    readLock.unlock(); // t = 3000
+
+    readLock.unlock(); // t = 3000
+    assertEquals(1, wlogged.get());
+    assertEquals(0, wsuppresed.get());
+    assertEquals(3000, totalHelpTime.get());
+  }
 }

hadoop-project/pom.xml

@@ -70,7 +70,7 @@
 
     <!-- jackson versions -->
     <jackson2.version>2.12.7</jackson2.version>
-    <jackson2.databind.version>2.12.7</jackson2.databind.version>
+    <jackson2.databind.version>2.12.7.1</jackson2.databind.version>
 
     <!-- httpcomponents versions -->
     <httpclient.version>4.5.13</httpclient.version>
@@ -117,7 +117,7 @@
     <commons-codec.version>1.15</commons-codec.version>
     <commons-collections.version>3.2.2</commons-collections.version>
     <commons-compress.version>1.21</commons-compress.version>
-    <commons-csv.version>1.0</commons-csv.version>
+    <commons-csv.version>1.9.0</commons-csv.version>
     <commons-io.version>2.11.0</commons-io.version>
     <commons-lang3.version>3.12.0</commons-lang3.version>
     <commons-logging.version>1.1.3</commons-logging.version>

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java

@@ -640,7 +640,10 @@ public static AWSCredentialProviderList buildAWSProviderList(
     AWSCredentialProviderList providers = new AWSCredentialProviderList();
     for (Class<?> aClass : awsClasses) {
 
-      if (aClass.getName().contains(AWS_AUTH_CLASS_PREFIX)) {
+      // List of V1 credential providers that will be migrated with V2 upgrade
+      if (!Arrays.asList("EnvironmentVariableCredentialsProvider",
+              "EC2ContainerCredentialsProviderWrapper", "InstanceProfileCredentialsProvider")
+          .contains(aClass.getSimpleName()) && aClass.getName().contains(AWS_AUTH_CLASS_PREFIX)) {
         V2Migration.v1ProviderReferenced(aClass.getName());
       }
 

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/s3_select.md

@@ -934,6 +934,21 @@ Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: GZIP is not applic
   ...
 ```
 
+
+### AWSBadRequestException  `UnsupportedStorageClass`
+
+S3 Select doesn't work with some storage classes like Glacier or Reduced Redundancy.
+Make sure you've set `fs.s3a.create.storage.class` to a supported storage class for S3 Select.
+
+```
+org.apache.hadoop.fs.s3a.AWSBadRequestException:
+    Select on s3a://example/dataset.csv.gz:
+    com.amazonaws.services.s3.model.AmazonS3Exception:
+     We do not support REDUCED_REDUNDANCY storage class.
+     Please check the service documentation and try again.
+     (Service: Amazon S3; Status Code: 400; Error Code: UnsupportedStorageClass
+```
+
 ### `PathIOException`: "seek() not supported"
 
 The input stream returned by the select call does not support seeking

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/AbstractS3SelectTest.java

@@ -60,7 +60,9 @@
 import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
 import org.apache.hadoop.util.DurationInfo;
 
+import static org.apache.hadoop.fs.s3a.Constants.STORAGE_CLASS;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.getLandsatCSVPath;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.select.CsvFile.ALL_QUOTES;
 import static org.apache.hadoop.fs.s3a.select.SelectConstants.*;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
@@ -280,6 +282,14 @@ boolean isSelectAvailable(final FileSystem filesystem) {
         .hasCapability(S3_SELECT_CAPABILITY);
   }
 
+  @Override
+  protected Configuration createConfiguration() {
+    Configuration conf = super.createConfiguration();
+    removeBaseAndBucketOverrides(conf, STORAGE_CLASS);
+
+    return conf;
+  }
+
   /**
    * Setup: requires select to be available.
    */

hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/SLSCapacityScheduler.java

@@ -36,6 +36,8 @@
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCommitRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
 import org.apache.hadoop.yarn.sls.SLSRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Private
 @Unstable
@@ -45,6 +47,7 @@ public class SLSCapacityScheduler extends CapacityScheduler implements
   private final SLSSchedulerCommons schedulerCommons;
   private Configuration conf;
   private SLSRunner runner;
+  private static final Logger LOG = LoggerFactory.getLogger(SLSCapacityScheduler.class);
 
   public SLSCapacityScheduler() {
     schedulerCommons = new SLSSchedulerCommons(this);
@@ -105,7 +108,12 @@ public boolean tryCommit(Resource cluster, ResourceCommitRequest r,
 
   @Override
   public void handle(SchedulerEvent schedulerEvent) {
-    schedulerCommons.handle(schedulerEvent);
+    try {
+      schedulerCommons.handle(schedulerEvent);
+    } catch(Exception e) {
+      LOG.error("Caught exception while handling scheduler event", e);
+      throw e;
+    }
   }
 
   @Override