HDFS-17509. RBF: Fix ClientProtocol.concat will throw NPE if tgr is a empty file.

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

@@ -667,39 +667,30 @@ public void rename2(final String src, final String dst,
   public void concat(String trg, String[] src) throws IOException {
     rpcServer.checkOperation(NameNode.OperationCategory.WRITE);
 
-    // See if the src and target files are all in the same namespace
-    LocatedBlocks targetBlocks = getBlockLocations(trg, 0, 1);
-    if (targetBlocks == null) {
-      throw new IOException("Cannot locate blocks for target file - " + trg);
-    }
-    LocatedBlock lastLocatedBlock = targetBlocks.getLastLocatedBlock();
-    String targetBlockPoolId = lastLocatedBlock.getBlock().getBlockPoolId();
-    for (String source : src) {
-      LocatedBlocks sourceBlocks = getBlockLocations(source, 0, 1);
-      if (sourceBlocks == null) {
-        throw new IOException(
-            "Cannot located blocks for source file " + source);
-      }
-      String sourceBlockPoolId =
-          sourceBlocks.getLastLocatedBlock().getBlock().getBlockPoolId();
-      if (!sourceBlockPoolId.equals(targetBlockPoolId)) {
-        throw new IOException("Cannot concatenate source file " + source
-            + " because it is located in a different namespace"
-            + " with block pool id " + sourceBlockPoolId
-            + " from the target file with block pool id "
-            + targetBlockPoolId);
-      }
+    // Concat only effects when all files in the same namespace.
+    RemoteResult<RemoteLocation, HdfsFileStatus> trgResult = getFileRemoteResult(trg);
+    if (trgResult.getResult() == null) {
+      throw new IOException("Cannot find target file - " + trg);
     }
+    RemoteLocation targetDestination = trgResult.getLocation();
+    String targetNameService = trgResult.getLocation().getNameserviceId();
 
-    // Find locations in the matching namespace.
-    final RemoteLocation targetDestination =
-        rpcServer.getLocationForPath(trg, true, targetBlockPoolId);
     String[] sourceDestinations = new String[src.length];
     for (int i = 0; i < src.length; i++) {
       String sourceFile = src[i];
-      RemoteLocation location =
-          rpcServer.getLocationForPath(sourceFile, true, targetBlockPoolId);
-      sourceDestinations[i] = location.getDest();
+      RemoteResult<RemoteLocation, HdfsFileStatus> srcResult = getFileRemoteResult(sourceFile);
+      if (srcResult.getResult() == null) {
+        throw new IOException("Cannot find source file - " + sourceFile);
+      }
+      RemoteLocation srcLocation = srcResult.getLocation();
+      sourceDestinations[i] = srcLocation.getDest();
+
+      if (!targetNameService.equals(srcLocation.getNameserviceId())) {
+        throw new IOException("Cannot concatenate source file " + sourceFile
+            + " because it is located in a different namespace" + " with nameservice "
+            + srcLocation.getNameserviceId() + " from the target file with nameservice "
+            + targetNameService);
+      }
     }
     // Invoke
     RemoteMethod method = new RemoteMethod("concat",
@@ -1009,6 +1000,20 @@ public HdfsFileStatus getFileInfo(String src) throws IOException {
     return ret;
   }
 
+  public RemoteResult<RemoteLocation, HdfsFileStatus> getFileRemoteResult(String path)
+      throws IOException {
+    rpcServer.checkOperation(NameNode.OperationCategory.READ);
+
+    final List<RemoteLocation> locations = rpcServer.getLocationsForPath(path, false, false);
+    RemoteMethod method =
+        new RemoteMethod("getFileInfo", new Class<?>[] {String.class}, new RemoteParam());
+    // Check for file information sequentially
+    RemoteResult<RemoteLocation, HdfsFileStatus> result =
+        rpcClient.invokeSequential(method, locations, HdfsFileStatus.class, null);
+
+    return result;
+  }
+
   @Override
   public boolean isFileClosed(String src) throws IOException {
     rpcServer.checkOperation(NameNode.OperationCategory.READ);

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

@@ -1697,42 +1697,6 @@ public Long getNextSPSPath() throws IOException {
     return nnProto.getNextSPSPath();
   }
 
-  /**
-   * Locate the location with the matching block pool id.
-   *
-   * @param path Path to check.
-   * @param failIfLocked Fail the request if locked (top mount point).
-   * @param blockPoolId The block pool ID of the namespace to search for.
-   * @return Prioritized list of locations in the federated cluster.
-   * @throws IOException if the location for this path cannot be determined.
-   */
-  protected RemoteLocation getLocationForPath(
-      String path, boolean failIfLocked, String blockPoolId)
-          throws IOException {
-
-    final List<RemoteLocation> locations =
-        getLocationsForPath(path, failIfLocked);
-
-    String nameserviceId = null;
-    Set<FederationNamespaceInfo> namespaces =
-        this.namenodeResolver.getNamespaces();
-    for (FederationNamespaceInfo namespace : namespaces) {
-      if (namespace.getBlockPoolId().equals(blockPoolId)) {
-        nameserviceId = namespace.getNameserviceId();
-        break;
-      }
-    }
-    if (nameserviceId != null) {
-      for (RemoteLocation location : locations) {
-        if (location.getNameserviceId().equals(nameserviceId)) {
-          return location;
-        }
-      }
-    }
-    throw new IOException(
-        "Cannot locate a nameservice for block pool " + blockPoolId);
-  }
-
   /**
    * Get the possible locations of a path in the federated cluster.
    * During the get operation, it will do the quota verification.

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

@@ -1224,6 +1224,17 @@ public void testProxyConcatFile() throws Exception {
     String badPath = "/unknownlocation/unknowndir";
     compareResponses(routerProtocol, nnProtocol, m,
         new Object[] {badPath, new String[] {routerFile}});
+
+    // Test when concat trg is a empty file
+    createFile(routerFS, existingFile, existingFileSize);
+    String sameRouterEmptyFile =
+        cluster.getFederatedTestDirectoryForNS(sameNameservice) +
+            "_newemptyfile";
+    createFile(routerFS, sameRouterEmptyFile, 0);
+    // Concat in same namespaces, succeeds
+    testConcat(existingFile, sameRouterEmptyFile, false);
+    FileStatus mergedStatus = getFileStatus(routerFS, sameRouterEmptyFile);
+    assertEquals(existingFileSize, mergedStatus.getLen());
   }
 
   @Test