Merge branch 'apache:trunk' into YARN-11262-PART5

Author: slfan1989

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

@@ -202,9 +202,24 @@ public static Object getExternalHandler() {
   private final boolean fallbackAllowed;
   private final boolean bindToWildCardAddress;
   private final byte[] clientId;
-  private final int maxAsyncCalls;
+  private int maxAsyncCalls;
   private final AtomicInteger asyncCallCounter = new AtomicInteger(0);
 
+  @VisibleForTesting
+  public int getAsyncCallCounter() {
+    return asyncCallCounter.get();
+  }
+
+  @VisibleForTesting
+  public void setMaxAsyncCalls(int limits) {
+    this.maxAsyncCalls = limits;
+  }
+
+  @VisibleForTesting
+  public boolean isAsyncCallCheckEabled() {
+    return maxAsyncCalls >= 0;
+  }
+
   /**
    * set the ping interval value in configuration
    * 
@@ -1460,7 +1475,7 @@ public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
   }
 
   private void checkAsyncCall() throws IOException {
-    if (isAsynchronousMode()) {
+    if (isAsynchronousMode() && isAsyncCallCheckEabled()) {
       if (asyncCallCounter.incrementAndGet() > maxAsyncCalls) {
         String errMsg = String.format(
             "Exceeded limit of max asynchronous calls: %d, " +
@@ -1518,7 +1533,7 @@ Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
         throw ioe;
       }
     } catch (Exception e) {
-      if (isAsynchronousMode()) {
+      if (isAsynchronousMode() && isAsyncCallCheckEabled()) {
         releaseAsyncCall();
       }
       throw e;
@@ -1527,7 +1542,9 @@ Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
     if (isAsynchronousMode()) {
       CompletableFuture<Writable> result = call.rpcResponseFuture.handle(
           (rpcResponse, e) -> {
-            releaseAsyncCall();
+            if (isAsyncCallCheckEabled()) {
+              releaseAsyncCall();
+            }
             if (e != null) {
               IOException ioe = (IOException) e;
               throw new CompletionException(warpIOException(ioe, connection));

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

@@ -27,12 +27,17 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.nio.file.FileSystems;
 import java.nio.file.Files;
+import java.nio.file.attribute.AclEntry;
+import java.nio.file.attribute.AclEntryPermission;
 import java.nio.file.attribute.FileAttribute;
-import java.nio.file.attribute.PosixFilePermission;
 import java.nio.file.attribute.PosixFilePermissions;
+import java.nio.file.attribute.UserPrincipal;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
 import java.util.Enumeration;
 import java.util.List;
 import java.util.Set;
@@ -51,6 +56,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.nio.file.attribute.AclEntryFlag.DIRECTORY_INHERIT;
+import static java.nio.file.attribute.AclEntryFlag.FILE_INHERIT;
+import static java.nio.file.attribute.AclEntryType.ALLOW;
+
 /** Run a Hadoop job jar. */
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
@@ -286,25 +295,19 @@ public void run(String[] args) throws Throwable {
     }
     mainClassName = mainClassName.replaceAll("/", ".");
 
-    File tmpDir = new File(System.getProperty("java.io.tmpdir"));
-    ensureDirectory(tmpDir);
-
     final File workDir;
     try {
-      FileAttribute<Set<PosixFilePermission>> perms = PosixFilePermissions
-          .asFileAttribute(PosixFilePermissions.fromString("rwx------"));
-      workDir = Files.createTempDirectory(tmpDir.toPath(), "hadoop-unjar", perms).toFile();
+      workDir = createWorkDirectory();
     } catch (IOException | SecurityException e) {
       // If user has insufficient perms to write to tmpDir, default
       // "Permission denied" message doesn't specify a filename.
       System.err.println("Error creating temp dir in java.io.tmpdir "
-                         + tmpDir + " due to " + e.getMessage());
+                         + System.getProperty("java.io.tmpdir") + " due to "
+                         + e.getMessage());
       System.exit(-1);
       return;
     }
 
-    ensureDirectory(workDir);
-
     ShutdownHookManager.get().addShutdownHook(
         new Runnable() {
           @Override
@@ -333,6 +336,55 @@ public void run() {
     }
   }
 
+  static File createWorkDirectory() throws IOException {
+    File tmpDir = new File(System.getProperty("java.io.tmpdir"));
+    ensureDirectory(tmpDir);
+
+    File workDir = Files.createTempDirectory(tmpDir.toPath(), "hadoop-unjar",
+        directoryPermissions()).toFile();
+    ensureDirectory(workDir);
+    return workDir;
+  }
+
+  private static FileAttribute<?> directoryPermissions() throws IOException {
+    Set<String> views = FileSystems.getDefault().supportedFileAttributeViews();
+    if (views.contains("posix")) {
+      return PosixFilePermissions
+          .asFileAttribute(PosixFilePermissions.fromString("rwx------"));
+    } else if (views.contains("acl")) {
+      return userOnly();
+    } else {
+      throw new IOException("unrecognized FileSystem type " +
+          FileSystems.getDefault());
+    }
+  }
+
+  private static FileAttribute<?> userOnly() throws IOException {
+    UserPrincipal user =
+        FileSystems.getDefault()
+            .getUserPrincipalLookupService()
+            .lookupPrincipalByName(System.getProperty("user.name"));
+    List<AclEntry> acl =
+        Collections.singletonList(AclEntry.newBuilder()
+            .setType(ALLOW)
+            .setPrincipal(user)
+            .setPermissions(EnumSet.allOf(AclEntryPermission.class))
+            .setFlags(DIRECTORY_INHERIT, FILE_INHERIT)
+            .build());
+    return
+        new FileAttribute<List<AclEntry>>() {
+          @Override
+          public String name() {
+            return "acl:acl";
+          }
+
+          @Override
+          public List<AclEntry> value() {
+            return acl;
+          }
+        };
+  }
+
   /**
    * Creates a classloader based on the environment that was specified by the
    * user. If HADOOP_USE_CLIENT_CLASSLOADER is specified, it creates an

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

@@ -2409,6 +2409,7 @@ The switch to turn S3A auditing on or off.
   <value>100</value>
   <description>
     Define the maximum number of outstanding async calls.
+    If negative, there is no limit on the number of outstanding async calls.
   </description>
 </property>
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAsyncIPC.java

@@ -38,7 +38,11 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -65,7 +69,7 @@ public void setupConf() {
     conf = new Configuration();
     conf.setInt(CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_MAX_KEY, 10000);
     Client.setPingInterval(conf, TestIPC.PING_INTERVAL);
-    // set asynchronous mode for main thread
+    // Set asynchronous mode for main thread.
     Client.setAsynchronousMode(true);
   }
 
@@ -78,17 +82,22 @@ static class AsyncCaller extends Thread {
         new HashMap<Integer, Future<LongWritable>>();
     Map<Integer, Long> expectedValues = new HashMap<Integer, Long>();
 
-    public AsyncCaller(Client client, InetSocketAddress server, int count) {
+    AsyncCaller(Client client, InetSocketAddress server, int count,
+        boolean checkAsyncCallEnabled) {
       this.client = client;
+      // Disable checkAsyncCall.
+      if (!checkAsyncCallEnabled) {
+        this.client.setMaxAsyncCalls(-1);
+      }
       this.server = server;
       this.count = count;
-      // set asynchronous mode, since AsyncCaller extends Thread
+      // Set asynchronous mode, since AsyncCaller extends Thread.
       Client.setAsynchronousMode(true);
     }
 
     @Override
     public void run() {
-      // in case Thread#Start is called, which will spawn new thread
+      // In case Thread#Start is called, which will spawn new thread.
       Client.setAsynchronousMode(true);
       for (int i = 0; i < count; i++) {
         try {
@@ -227,7 +236,7 @@ public AsyncLimitlCaller(int callerId, Client client, InetSocketAddress server,
       this.client = client;
       this.server = server;
       this.count = count;
-      // set asynchronous mode, since AsyncLimitlCaller extends Thread
+      // Set asynchronous mode, since AsyncLimitlCaller extends Thread.
       Client.setAsynchronousMode(true);
       this.callerId = callerId;
     }
@@ -285,12 +294,19 @@ private void waitForReturnValues(final int start, final int end)
     }
   }
 
+  @Test
+  @Timeout(value = 60)
+  public void testAsyncCallCheckDisabled() throws IOException, InterruptedException,
+      ExecutionException {
+    internalTestAsyncCall(3, true, 2, 5, 10, false);
+  }
+
   @Test
   @Timeout(value = 60)
   public void testAsyncCall() throws IOException, InterruptedException,
       ExecutionException {
-    internalTestAsyncCall(3, false, 2, 5, 100);
-    internalTestAsyncCall(3, true, 2, 5, 10);
+    internalTestAsyncCall(3, false, 2, 5, 100, true);
+    internalTestAsyncCall(3, true, 2, 5, 10, true);
   }
 
   @Test
@@ -301,7 +317,8 @@ public void testAsyncCallLimit() throws IOException,
   }
 
   public void internalTestAsyncCall(int handlerCount, boolean handlerSleep,
-      int clientCount, int callerCount, int callCount) throws IOException,
+      int clientCount, int callerCount, int callCount,
+      boolean checkAsyncCallEnabled) throws IOException,
       InterruptedException, ExecutionException {
     Server server = new TestIPC.TestServer(handlerCount, handlerSleep, conf);
     InetSocketAddress addr = NetUtils.getConnectAddress(server);
@@ -314,10 +331,14 @@ public void internalTestAsyncCall(int handlerCount, boolean handlerSleep,
 
     AsyncCaller[] callers = new AsyncCaller[callerCount];
     for (int i = 0; i < callerCount; i++) {
-      callers[i] = new AsyncCaller(clients[i % clientCount], addr, callCount);
+      callers[i] = new AsyncCaller(clients[i % clientCount], addr, callCount,
+          checkAsyncCallEnabled);
       callers[i].start();
     }
     for (int i = 0; i < callerCount; i++) {
+      if (!checkAsyncCallEnabled) {
+        assertEquals(0, clients[i % clientCount].getAsyncCallCounter());
+      }
       callers[i].join();
       callers[i].assertReturnValues();
     }
@@ -340,7 +361,7 @@ public void testCallGetReturnRpcResponseMultipleTimes() throws IOException,
     int asyncCallCount = client.getAsyncCallCount();
 
     try {
-      AsyncCaller caller = new AsyncCaller(client, addr, callCount);
+      AsyncCaller caller = new AsyncCaller(client, addr, callCount, true);
       caller.run();
       caller.assertReturnValues();
       caller.assertReturnValues();
@@ -364,7 +385,7 @@ public void testFutureGetWithTimeout() throws IOException,
     final Client client = new Client(LongWritable.class, conf);
 
     try {
-      final AsyncCaller caller = new AsyncCaller(client, addr, 10);
+      final AsyncCaller caller = new AsyncCaller(client, addr, 10, true);
       caller.run();
       caller.assertReturnValues(10, TimeUnit.MILLISECONDS);
     } finally {
@@ -463,7 +484,7 @@ public void run() {
     try {
       InetSocketAddress addr = NetUtils.getConnectAddress(server);
       server.start();
-      final AsyncCaller caller = new AsyncCaller(client, addr, 4);
+      final AsyncCaller caller = new AsyncCaller(client, addr, 4, true);
       caller.run();
       caller.assertReturnValues();
     } finally {
@@ -501,7 +522,7 @@ public void run() {
     try {
       InetSocketAddress addr = NetUtils.getConnectAddress(server);
       server.start();
-      final AsyncCaller caller = new AsyncCaller(client, addr, 10);
+      final AsyncCaller caller = new AsyncCaller(client, addr, 10, true);
       caller.run();
       caller.assertReturnValues();
     } finally {
@@ -538,7 +559,7 @@ public void run() {
     try {
       InetSocketAddress addr = NetUtils.getConnectAddress(server);
       server.start();
-      final AsyncCaller caller = new AsyncCaller(client, addr, 10);
+      final AsyncCaller caller = new AsyncCaller(client, addr, 10, true);
       caller.run();
       caller.assertReturnValues();
     } finally {
@@ -580,7 +601,7 @@ public void run() {
       server.start();
       AsyncCaller[] callers = new AsyncCaller[callerCount];
       for (int i = 0; i < callerCount; ++i) {
-        callers[i] = new AsyncCaller(client, addr, perCallerCallCount);
+        callers[i] = new AsyncCaller(client, addr, perCallerCallCount, true);
         callers[i].start();
       }
       for (int i = 0; i < callerCount; ++i) {

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

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.util.RunJar.MATCH_ANY;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -197,6 +198,24 @@ private File getUnjarDir(String dirName) {
     return unjarDir;
   }
 
+  /**
+   * Tests the creation of the temp working directory into which the jars are
+   * unjarred.
+   */
+  @Test
+  public void testCreateWorkDirectory() throws Exception {
+    File workDir = null;
+    try {
+      workDir = RunJar.createWorkDirectory();
+
+      assertNotNull(workDir, "Work directory should exist and not null");
+    } finally {
+      if (workDir != null) {
+        FileUtil.fullyDelete(workDir);
+      }
+    }
+  }
+
   /**
    * Tests the client classloader to verify the main class and its dependent
    * class are loaded correctly by the application classloader, and others are

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/fs/contract/router/RouterHDFSContract.java

@@ -27,7 +27,7 @@
 import org.apache.hadoop.fs.contract.hdfs.HDFSContract;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster;
-import org.junit.Assert;
+import org.junit.jupiter.api.Assertions;
 
 /**
  * The contract of Router-based Federated HDFS.
@@ -109,7 +109,7 @@ public static MiniRouterDFSCluster getRouterCluster() {
 
   public static FileSystem getFileSystem() throws IOException {
     //assumes cluster is not null
-    Assert.assertNotNull("cluster not created", cluster);
+    Assertions.assertNotNull(cluster, "cluster not created");
     return cluster.getRandomRouter().getFileSystem();
   }