HDFS-16865. The source path is always / after RBF proxied the complete, addBlock and getAdditionalDatanode RPC. (#5200). Contributed by ZanderXu.

Reviewed-by: Inigo Goiri <inigoiri@apache.org>
Signed-off-by: Ayush Saxena <ayushsaxena@apache.org>

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java

@@ -483,12 +483,10 @@ public class RouterClientProtocol implements ClientProtocol {
         new RemoteParam(), clientName, previous, excludedNodes, fileId,
         favoredNodes, addBlockFlags);
 
+    final List<RemoteLocation> locations = rpcServer.getLocationsForPath(src, true);
     if (previous != null) {
-      return rpcClient.invokeSingle(previous, method, LocatedBlock.class);
+      return rpcClient.invokeSingle(previous, method, locations, LocatedBlock.class);
     }
-
-    final List<RemoteLocation> locations =
-        rpcServer.getLocationsForPath(src, true);
     // TODO verify the excludedNodes and favoredNodes are acceptable to this NN
     return rpcClient.invokeSequential(
         locations, method, LocatedBlock.class, null);
@@ -513,12 +511,11 @@ public class RouterClientProtocol implements ClientProtocol {
         new RemoteParam(), fileId, blk, existings, existingStorageIDs, excludes,
         numAdditionalNodes, clientName);
 
+    final List<RemoteLocation> locations = rpcServer.getLocationsForPath(src, false);
     if (blk != null) {
-      return rpcClient.invokeSingle(blk, method, LocatedBlock.class);
+      return rpcClient.invokeSingle(blk, method, locations, LocatedBlock.class);
     }
 
-    final List<RemoteLocation> locations =
-        rpcServer.getLocationsForPath(src, false);
     return rpcClient.invokeSequential(
         locations, method, LocatedBlock.class, null);
   }
@@ -532,7 +529,9 @@ public class RouterClientProtocol implements ClientProtocol {
         new Class<?>[] {ExtendedBlock.class, long.class, String.class,
             String.class},
         b, fileId, new RemoteParam(), holder);
-    rpcClient.invokeSingle(b, method);
+
+    final List<RemoteLocation> locations = rpcServer.getLocationsForPath(src, false);
+    rpcClient.invokeSingle(b, method, locations, Void.class);
   }
 
   @Override
@@ -545,12 +544,11 @@ public class RouterClientProtocol implements ClientProtocol {
             long.class},
         new RemoteParam(), clientName, last, fileId);
 
+    final List<RemoteLocation> locations = rpcServer.getLocationsForPath(src, true);
     if (last != null) {
-      return rpcClient.invokeSingle(last, method, Boolean.class);
+      return rpcClient.invokeSingle(last, method, locations, Boolean.class);
     }
 
-    final List<RemoteLocation> locations =
-        rpcServer.getLocationsForPath(src, true);
     // Complete can return true/false, so don't expect a result
     return rpcClient.invokeSequential(locations, method, Boolean.class, null);
   }
@@ -580,7 +578,7 @@ public class RouterClientProtocol implements ClientProtocol {
         new Class<?>[] {String.class, ExtendedBlock.class, ExtendedBlock.class,
             DatanodeID[].class, String[].class},
         clientName, oldBlock, newBlock, newNodes, newStorageIDs);
-    rpcClient.invokeSingle(oldBlock, method);
+    rpcClient.invokeSingleBlockPool(oldBlock.getBlockPoolId(), method);
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java

@@ -848,6 +848,26 @@ public class RouterRpcClient {
     return ret;
   }
 
+  /**
+   * Try to get the remote location whose bpId is same with the input bpId from the input locations.
+   * @param locations the input RemoteLocations.
+   * @param bpId the input bpId.
+   * @return the remote location whose bpId is same with the input.
+   * @throws IOException
+   */
+  private RemoteLocation getLocationWithBPID(List<RemoteLocation> locations, String bpId)
+      throws IOException {
+    String nsId = getNameserviceForBlockPoolId(bpId);
+    for (RemoteLocation l : locations) {
+      if (l.getNameserviceId().equals(nsId)) {
+        return l;
+      }
+    }
+
+    LOG.debug("Can't find remote location for the {} from {}", bpId, locations);
+    return new RemoteLocation(nsId, "/", "/");
+  }
+
   /**
    * Invokes a ClientProtocol method. Determines the target nameservice via a
    * provided block.
@@ -857,13 +877,15 @@ public class RouterRpcClient {
    *
    * @param block Block used to determine appropriate nameservice.
    * @param method The remote method and parameters to invoke.
+   * @param locations The remote locations will be used.
+   * @param clazz – Class for the return type.
    * @return The result of invoking the method.
    * @throws IOException If the invoke generated an error.
    */
-  public Object invokeSingle(final ExtendedBlock block, RemoteMethod method)
-      throws IOException {
-    String bpId = block.getBlockPoolId();
-    return invokeSingleBlockPool(bpId, method);
+  public <T> T invokeSingle(final ExtendedBlock block, RemoteMethod method,
+      final List<RemoteLocation> locations, Class<T> clazz) throws IOException {
+    RemoteLocation location = getLocationWithBPID(locations, block.getBlockPoolId());
+    return invokeSingle(location, method, clazz);
   }
 
   /**

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

@@ -73,6 +73,7 @@ import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ipc.StandbyException;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.junit.Test;
+import org.slf4j.event.Level;
 
 /**
  * The RPC interface of the {@link getRouter()} implemented by
@@ -275,6 +276,14 @@ public class TestRouterRpcMultiDestination extends TestRouterRpc {
   @Test
   public void testPreviousBlockNotNull()
       throws IOException, URISyntaxException {
+    final GenericTestUtils.LogCapturer stateChangeLog =
+        GenericTestUtils.LogCapturer.captureLogs(NameNode.stateChangeLog);
+    GenericTestUtils.setLogLevel(NameNode.stateChangeLog, Level.DEBUG);
+
+    final GenericTestUtils.LogCapturer nameNodeLog =
+        GenericTestUtils.LogCapturer.captureLogs(NameNode.LOG);
+    GenericTestUtils.setLogLevel(NameNode.LOG, Level.DEBUG);
+
     final FederationRPCMetrics metrics = getRouterContext().
         getRouter().getRpcServer().getRPCMetrics();
     final ClientProtocol clientProtocol = getRouterProtocol();
@@ -305,6 +314,7 @@ public class TestRouterRpcMultiDestination extends TestRouterRpc {
       long proxyNumAddBlock = metrics.getProcessingOps();
       assertEquals(2, proxyNumAddBlock - proxyNumCreate);
 
+      stateChangeLog.clearOutput();
       // Add a block via router and previous block is not null.
       LocatedBlock blockTwo = clientProtocol.addBlock(
           testPath, clientName, blockOne.getBlock(), null,
@@ -312,7 +322,9 @@ public class TestRouterRpcMultiDestination extends TestRouterRpc {
       assertNotNull(blockTwo);
       long proxyNumAddBlock2 = metrics.getProcessingOps();
       assertEquals(1, proxyNumAddBlock2 - proxyNumAddBlock);
+      assertTrue(stateChangeLog.getOutput().contains("BLOCK* getAdditionalBlock: " + testPath));
 
+      nameNodeLog.clearOutput();
       // Get additionalDatanode via router and block is not null.
       DatanodeInfo[] exclusions = DatanodeInfo.EMPTY_ARRAY;
       LocatedBlock newBlock = clientProtocol.getAdditionalDatanode(
@@ -322,12 +334,15 @@ public class TestRouterRpcMultiDestination extends TestRouterRpc {
       assertNotNull(newBlock);
       long proxyNumAdditionalDatanode = metrics.getProcessingOps();
       assertEquals(1, proxyNumAdditionalDatanode - proxyNumAddBlock2);
+      assertTrue(nameNodeLog.getOutput().contains("getAdditionalDatanode: src=" + testPath));
 
+      stateChangeLog.clearOutput();
       // Complete the file via router and last block is not null.
       clientProtocol.complete(testPath, clientName,
           newBlock.getBlock(), status.getFileId());
       long proxyNumComplete = metrics.getProcessingOps();
       assertEquals(1, proxyNumComplete - proxyNumAdditionalDatanode);
+      assertTrue(stateChangeLog.getOutput().contains("DIR* NameSystem.completeFile: " + testPath));
     } finally {
       clientProtocol.delete(testPath, true);
     }