Merge branch 'trunk' of https://github.com/ABFSDriver/AbfsHadoop into HADOOP-19497

Author: bhattmanish98

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

@@ -1462,7 +1462,6 @@ public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
   private void checkAsyncCall() throws IOException {
     if (isAsynchronousMode()) {
       if (asyncCallCounter.incrementAndGet() > maxAsyncCalls) {
-        asyncCallCounter.decrementAndGet();
         String errMsg = String.format(
             "Exceeded limit of max asynchronous calls: %d, " +
             "please configure %s to adjust it.",
@@ -1518,7 +1517,7 @@ Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
         ioe.initCause(ie);
         throw ioe;
       }
-    } catch(Exception e) {
+    } catch (Exception e) {
       if (isAsynchronousMode()) {
         releaseAsyncCall();
       }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MultipleDestinationMountTableResolver.java

@@ -118,4 +118,9 @@ public PathLocation getDestinationForPath(String path) throws IOException {
   public void addResolver(DestinationOrder order, OrderedResolver resolver) {
     orderedResolvers.put(order, resolver);
   }
-}
+
+  @VisibleForTesting
+  public OrderedResolver getOrderedResolver(DestinationOrder order) {
+    return orderedResolvers.get(order);
+  }
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/order/LocalResolver.java

@@ -45,6 +45,7 @@
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.net.HostAndPort;
 
+import static org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil.syncReturn;
 
 /**
  * The local subcluster (where the writer is) should be tried first. The writer
@@ -124,9 +125,9 @@ String getClientAddr() {
    * needs to be done as a privileged action to use the user for the Router and
    * not the one from the client in the RPC call.
    *
-   * @return DN IP -> Subcluster.
+   * @return DN IP -> Subcluster.
    */
-  private Map<String, String> getDatanodesSubcluster() {
+  public Map<String, String> getDatanodesSubcluster() {
 
     final RouterRpcServer rpcServer = getRpcServer();
     if (rpcServer == null) {
@@ -143,9 +144,16 @@ private Map<String, String> getDatanodesSubcluster() {
             @Override
             public Map<String, DatanodeStorageReport[]> run() {
               try {
-                return rpcServer.getDatanodeStorageReportMap(
-                    DatanodeReportType.ALL);
-              } catch (IOException e) {
+                Map<String, DatanodeStorageReport[]> result;
+                if (rpcServer.isAsync()) {
+                  rpcServer.getDatanodeStorageReportMapAsync(DatanodeReportType.ALL);
+                  result = syncReturn(Map.class);
+                } else {
+                  result = rpcServer.getDatanodeStorageReportMap(
+                      DatanodeReportType.ALL);
+                }
+                return result;
+              } catch (Exception e) {
                 LOG.error("Cannot get the datanodes from the RPC server", e);
                 return null;
               }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/utils/AsyncUtil.java

@@ -127,7 +127,7 @@ public static <R> R syncReturn(Class<R> clazz)
     try {
       return (R) completableFuture.get();
     } catch (ExecutionException e) {
-      throw (Exception)e.getCause();
+      throw (Exception) e.getCause();
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/TestRouterAsyncRPCMultipleDestinationMountTableResolver.java

@@ -25,16 +25,22 @@
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
 import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
 import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver;
 import org.apache.hadoop.hdfs.server.federation.resolver.MultipleDestinationMountTableResolver;
 import org.apache.hadoop.hdfs.server.federation.resolver.order.DestinationOrder;
+import org.apache.hadoop.hdfs.server.federation.resolver.order.LocalResolver;
 import org.apache.hadoop.hdfs.server.federation.router.RemoteMethod;
+import org.apache.hadoop.hdfs.server.federation.router.RouterClient;
 import org.apache.hadoop.hdfs.server.federation.router.RouterClientProtocol;
 import org.apache.hadoop.hdfs.server.federation.router.RouterQuotaUsage;
 import org.apache.hadoop.hdfs.server.federation.router.TestRouterRPCMultipleDestinationMountTableResolver;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest;
 import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -52,14 +58,21 @@
 public class TestRouterAsyncRPCMultipleDestinationMountTableResolver extends
     TestRouterRPCMultipleDestinationMountTableResolver {
 
+  public static final Logger LOG =
+      LoggerFactory.getLogger(TestRouterAsyncRPCMultipleDestinationMountTableResolver.class);
+
   @BeforeClass
   public static void setUp() throws Exception {
 
     // Build and start a federated cluster.
     cluster = new StateStoreDFSCluster(false, 3,
         MultipleDestinationMountTableResolver.class);
-    Configuration routerConf =
-        new RouterConfigBuilder().stateStore().admin().quota().rpc().build();
+    Configuration routerConf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .quota()
+        .rpc()
+        .build();
     routerConf.setBoolean(DFS_ROUTER_ASYNC_RPC_ENABLE_KEY, true);
 
     Configuration hdfsConf = new Configuration(false);
@@ -84,6 +97,43 @@ public static void setUp() throws Exception {
     rpcServer =routerContext.getRouter().getRpcServer();
   }
 
+  @Test
+  public void testLocalResolverGetDatanodesSubcluster() throws IOException {
+    String testPath = "/testLocalResolverGetDatanodesSubcluster";
+    Path path = new Path(testPath);
+    Map<String, String> destMap = new HashMap<>();
+    destMap.put("ns0", testPath);
+    destMap.put("ns1", testPath);
+    nnFs0.mkdirs(path);
+    nnFs1.mkdirs(path);
+    MountTable addEntry =
+        MountTable.newInstance(testPath, destMap);
+    addEntry.setQuota(new RouterQuotaUsage.Builder().build());
+    addEntry.setDestOrder(DestinationOrder.LOCAL);
+    assertTrue(addMountTable(addEntry));
+
+    Map<String, String> datanodesSubcluster = null;
+    try {
+      MultipleDestinationMountTableResolver resolver =
+          (MultipleDestinationMountTableResolver) routerContext.getRouter().getSubclusterResolver();
+      LocalResolver localResolver =
+          (LocalResolver) resolver.getOrderedResolver(DestinationOrder.LOCAL);
+      datanodesSubcluster = localResolver.getDatanodesSubcluster();
+    } catch (Exception e) {
+      LOG.info("Exception occurs when testLocalResolverGetDatanodesSubcluster.", e);
+    } finally {
+      RouterClient client = routerContext.getAdminClient();
+      MountTableManager mountTableManager = client.getMountTableManager();
+      RemoveMountTableEntryRequest req2 =
+          RemoveMountTableEntryRequest.newInstance(testPath);
+      mountTableManager.removeMountTableEntry(req2);
+      nnFs0.delete(new Path(testPath), true);
+      nnFs1.delete(new Path(testPath), true);
+    }
+    assertNotNull(datanodesSubcluster);
+    assertFalse(datanodesSubcluster.isEmpty());
+  }
+
   @Override
   @Test
   public void testInvokeAtAvailableNs() throws IOException {

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/java/org/apache/hadoop/mapred/TestShuffleChannelHandler.java

@@ -44,6 +44,7 @@
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import io.netty.util.ReferenceCounted;
 import io.netty.util.concurrent.GlobalEventExecutor;
 
 import java.io.ByteArrayOutputStream;
@@ -61,6 +62,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import javax.crypto.SecretKey;
@@ -115,6 +117,7 @@ public void testGetMapsChunkedFileSSl() throws Exception {
     final LinkedList<Object> unencryptedMessages = new LinkedList<>();
     final EmbeddedChannel shuffle = t.createShuffleHandlerSSL(unencryptedMessages);
     t.testGetAllAttemptsForReduce0NoKeepAlive(unencryptedMessages, shuffle);
+    drainChannel(shuffle);
   }
 
   @Test
@@ -192,8 +195,10 @@ public void testIncompatibleShuffleVersion() {
 
     assertEquals(getExpectedHttpResponse(HttpResponseStatus.BAD_REQUEST).toString(),
         actual.toString());
+    tryRelease(actual);
 
     assertFalse(shuffle.isActive(), "closed"); // known-issue
+    drainChannel(decoder);
   }
 
   @Test
@@ -210,11 +215,13 @@ public void testInvalidMapNoIndexFile() {
     }
 
     DefaultHttpResponse actual = decoder.readInbound();
+    drainChannel(decoder);
     assertFalse(actual.headers().get(CONTENT_LENGTH).isEmpty());
     actual.headers().set(CONTENT_LENGTH, 0);
 
     assertEquals(getExpectedHttpResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR).toString(),
         actual.toString());
+    tryRelease(actual);
 
     assertFalse(shuffle.isActive(), "closed");
   }
@@ -237,15 +244,36 @@ public void testInvalidMapNoDataFile() {
     }
 
     DefaultHttpResponse actual = decoder.readInbound();
+    drainChannel(decoder);
     assertFalse(actual.headers().get(CONTENT_LENGTH).isEmpty());
     actual.headers().set(CONTENT_LENGTH, 0);
 
     assertEquals(getExpectedHttpResponse(HttpResponseStatus.INTERNAL_SERVER_ERROR).toString(),
         actual.toString());
+    tryRelease(actual);
 
     assertFalse(shuffle.isActive(), "closed");
   }
 
+  private void drainChannel(EmbeddedChannel ch) {
+    Object o;
+    while((o = ch.readInbound())!=null) {
+      tryRelease(o);
+    }
+    while((o = ch.readOutbound())!=null) {
+      tryRelease(o);
+    }
+  }
+
+  private void tryRelease(Object obj) {
+    if (obj instanceof ReferenceCounted) {
+      ReferenceCounted bb = (ReferenceCounted) obj;
+      if (bb.refCnt() > 0) {
+        bb.release(bb.refCnt());
+      }
+    }
+  }
+
   private DefaultHttpResponse getExpectedHttpResponse(HttpResponseStatus status) {
     DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, status);
     response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
@@ -365,8 +393,8 @@ private void testGetAllAttemptsForReduce0NoKeepAlive(
       assertFalse(shuffle.isActive(), "no keep-alive");
     }
 
-    private void testKeepAlive(java.util.Queue<Object> messages,
-                               EmbeddedChannel shuffle) throws IOException {
+    private void testKeepAlive(java.util.Queue<Object> messages, EmbeddedChannel shuffle)
+        throws IOException, InterruptedException, ExecutionException {
       final FullHttpRequest req1 = createRequest(
           getUri(TEST_JOB_ID, 0, Collections.singletonList(TEST_ATTEMPT_1), true));
       shuffle.writeInbound(req1);
@@ -375,6 +403,7 @@ private void testKeepAlive(java.util.Queue<Object> messages,
           getAttemptData(new Attempt(TEST_ATTEMPT_1, TEST_DATA_A))
       );
       assertTrue(shuffle.isActive(), "keep-alive");
+      drainChannel(shuffle);
       messages.clear();
 
       final FullHttpRequest req2 = createRequest(
@@ -385,6 +414,7 @@ private void testKeepAlive(java.util.Queue<Object> messages,
           getAttemptData(new Attempt(TEST_ATTEMPT_2, TEST_DATA_B))
       );
       assertTrue(shuffle.isActive(), "keep-alive");
+      drainChannel(shuffle);
       messages.clear();
 
       final FullHttpRequest req3 = createRequest(
@@ -395,6 +425,7 @@ private void testKeepAlive(java.util.Queue<Object> messages,
           getAttemptData(new Attempt(TEST_ATTEMPT_3, TEST_DATA_C))
       );
       assertFalse(shuffle.isActive(), "no keep-alive");
+      drainChannel(shuffle);
     }
 
     private ArrayList<ByteBuf> getAllAttemptsForReduce0() throws IOException {
@@ -431,11 +462,13 @@ private void assertResponse(java.util.Queue<Object> outboundMessages,
         decodeChannel.writeInbound(actualBytes);
         Object obj = decodeChannel.readInbound();
         LOG.info("Decoded object: {}", obj);
+        drainChannel(decodeChannel);
 
         if (i == 0) {
           DefaultHttpResponse resp = (DefaultHttpResponse) obj;
           assertEquals(response.toString(), resp.toString());
         }
+        tryRelease(obj);
         if (i > 0 && i <= content.size()) {
           assertEquals(ByteBufUtil.prettyHexDump(content.get(i - 1)),
               actualHexdump, "data should match");

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/java/org/apache/hadoop/mapred/TestShuffleHandlerBase.java

@@ -84,6 +84,13 @@ public void setup() throws IOException {
 
   @AfterEach
   public void teardown() {
+    //Trigger GC so that we get the leak warnings early
+    System.gc();
+    try {
+      // Wait for logger to flush
+      Thread.sleep(1000);
+    } catch (InterruptedException e) {
+    }
     System.setOut(standardOut);
     System.out.print(outputStreamCaptor);
     // For this to work ch.qos.logback.classic is needed for some reason