HDFS-16283. RBF: reducing the load of renewLease() RPC

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -41,6 +41,7 @@
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
@@ -188,6 +189,7 @@
 import org.apache.hadoop.util.DataChecksum.Type;
 import org.apache.hadoop.util.Lists;
 import org.apache.hadoop.util.Progressable;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Time;
 import org.apache.hadoop.tracing.TraceScope;
 import org.apache.hadoop.tracing.Tracer;
@@ -579,6 +581,28 @@ void updateLastLeaseRenewal() {
     }
   }
 
+  /**
+   * Get all nsIdentifies of DFSOutputStreams.
+   */
+  private String getRenewLeaseNSIdentifies() {
+    HashSet<String> allNSIdentifies = new HashSet<>();
+    synchronized (filesBeingWritten) {
+      if (filesBeingWritten.isEmpty()) {
+        return null;
+      }
+      for (DFSOutputStream outputStream : filesBeingWritten.values()) {
+        String nsIdentify = outputStream.getNsIdentify();
+        if (nsIdentify != null && !nsIdentify.isEmpty()) {
+          allNSIdentifies.add(nsIdentify);
+        }
+      }
+      if (allNSIdentifies.isEmpty()) {
+        return null;
+      }
+    }
+    return StringUtils.join(",", allNSIdentifies);
+  }
+
   /**
    * Renew leases.
    * @return true if lease was renewed. May return false if this
@@ -587,7 +611,7 @@ void updateLastLeaseRenewal() {
   public boolean renewLease() throws IOException {
     if (clientRunning && !isFilesBeingWrittenEmpty()) {
       try {
-        namenode.renewLease(clientName);
+        namenode.renewLease(clientName, getRenewLeaseNSIdentifies());
         updateLastLeaseRenewal();
         return true;
       } catch (IOException e) {

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java

@@ -113,6 +113,7 @@ public class DFSOutputStream extends FSOutputSummer
 
   protected final String src;
   protected final long fileId;
+  private final String nsIdentify;
   protected final long blockSize;
   protected final int bytesPerChecksum;
 
@@ -195,6 +196,7 @@ private DFSOutputStream(DFSClient dfsClient, String src,
     this.dfsClient = dfsClient;
     this.src = src;
     this.fileId = stat.getFileId();
+    this.nsIdentify = stat.getNsIdentify();
     this.blockSize = stat.getBlockSize();
     this.blockReplication = stat.getReplication();
     this.fileEncryptionInfo = stat.getFileEncryptionInfo();
@@ -1084,6 +1086,11 @@ public long getFileId() {
     return fileId;
   }
 
+  @VisibleForTesting
+  public String getNsIdentify() {
+    return nsIdentify;
+  }
+
   /**
    * Return the source of stream.
    */

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -763,7 +763,7 @@ SnapshotStatus[] getSnapshotListing(String snapshotRoot)
    * @throws IOException If an I/O error occurred
    */
   @Idempotent
-  void renewLease(String clientName) throws IOException;
+  void renewLease(String clientName, String allNSIdentifies) throws IOException;
 
   /**
    * Start lease recovery.

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java

@@ -490,6 +490,10 @@ default FileStatus makeQualified(URI defaultUri, Path parent) {
    */
   int compareTo(FileStatus stat);
 
+  void setNsIdentify(String nsIdentify);
+
+  String getNsIdentify();
+
   /**
    * Set redundant flags for compatibility with existing applications.
    */

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsLocatedFileStatus.java

@@ -54,6 +54,8 @@ public class HdfsLocatedFileStatus
   // BlockLocations[] is the user-facing type
   private transient LocatedBlocks hdfsloc;
 
+  private String nsIdentify = null;
+
   /**
    * Constructor.
    * @param length the number of bytes the file has
@@ -217,4 +219,14 @@ public LocatedFileStatus makeQualifiedLocated(URI defaultUri, Path path) {
     return this;
   }
 
+  @Override
+  public String getNsIdentify() {
+    return nsIdentify;
+  }
+
+  @Override
+  public void setNsIdentify(String nsIdentify) {
+    this.nsIdentify = nsIdentify;
+  }
+
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsNamedFileStatus.java

@@ -44,6 +44,8 @@ public class HdfsNamedFileStatus extends FileStatus implements HdfsFileStatus {
   private final int childrenNum;
   private final byte storagePolicy;
 
+  private String nsIdentify = null;
+
   /**
    * Constructor.
    * @param length the number of bytes the file has
@@ -177,4 +179,13 @@ public int hashCode() {
     return super.hashCode();
   }
 
+  @Override
+  public String getNsIdentify() {
+    return nsIdentify;
+  }
+
+  @Override
+  public void setNsIdentify(String nsIdentify) {
+    this.nsIdentify = nsIdentify;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java

@@ -744,11 +744,15 @@ public BatchedDirectoryListing getBatchedListing(
 
 
   @Override
-  public void renewLease(String clientName) throws IOException {
-    RenewLeaseRequestProto req = RenewLeaseRequestProto.newBuilder()
-        .setClientName(clientName).build();
+  public void renewLease(String clientName, String nsIdentifies)
+      throws IOException {
+    RenewLeaseRequestProto.Builder builder = RenewLeaseRequestProto
+        .newBuilder().setClientName(clientName);
+    if (nsIdentifies != null && !nsIdentifies.isEmpty()) {
+      builder.setNsIdentifies(nsIdentifies);
+    }
     try {
-      rpcProxy.renewLease(null, req);
+      rpcProxy.renewLease(null, builder.build());
     } catch (ServiceException e) {
       throw ProtobufHelper.getRemoteException(e);
     }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java

@@ -1764,7 +1764,7 @@ public static HdfsFileStatus convert(HdfsFileStatusProto fs) {
     EnumSet<HdfsFileStatus.Flags> flags = fs.hasFlags()
         ? convertFlags(fs.getFlags())
         : convertFlags(fs.getPermission());
-    return new HdfsFileStatus.Builder()
+    HdfsFileStatus hdfsFileStatus = new HdfsFileStatus.Builder()
         .length(fs.getLength())
         .isdir(fs.getFileType().equals(FileType.IS_DIR))
         .replication(fs.getBlockReplication())
@@ -1794,6 +1794,10 @@ public static HdfsFileStatus convert(HdfsFileStatusProto fs) {
             ? convertErasureCodingPolicy(fs.getEcPolicy())
             : null)
         .build();
+    if (fs.hasNsIdentify()) {
+      hdfsFileStatus.setNsIdentify(fs.getNsIdentify());
+    }
+    return hdfsFileStatus;
   }
 
   private static EnumSet<HdfsFileStatus.Flags> convertFlags(int flags) {
@@ -2399,6 +2403,9 @@ public static HdfsFileStatusProto convert(HdfsFileStatus fs) {
     flags |= fs.isSnapshotEnabled() ? HdfsFileStatusProto.Flags
         .SNAPSHOT_ENABLED_VALUE : 0;
     builder.setFlags(flags);
+    if (fs.getNsIdentify() != null && !fs.getNsIdentify().isEmpty()) {
+      builder.setNsIdentify(fs.getNsIdentify());
+    }
     return builder.build();
   }
 

hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/ClientNamenodeProtocol.proto

@@ -332,6 +332,7 @@ message GetSnapshotDiffReportListingResponseProto {
 }
 message RenewLeaseRequestProto {
   required string clientName = 1;
+  optional string nsIdentifies = 2;
 }
 
 message RenewLeaseResponseProto { //void response

hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/hdfs.proto

@@ -481,6 +481,7 @@ message HdfsFileStatusProto {
 
   // Set of flags
   optional uint32 flags = 18 [default = 0];
+  optional string nsIdentify = 19;
 }
 
 /**

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

@@ -114,6 +114,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -129,6 +130,8 @@ public class RouterClientProtocol implements ClientProtocol {
   private final RouterFederationRename rbfRename;
   private final FileSubclusterResolver subclusterResolver;
   private final ActiveNamenodeResolver namenodeResolver;
+  private final Map<String, FederationNamespaceInfo> nsNameSpaceInfoCache
+      = new ConcurrentHashMap<>();
 
   /**
    * Caching server defaults so as to prevent redundant calls to namenode,
@@ -291,8 +294,10 @@ public HdfsFileStatus create(String src, FsPermission masked,
     RemoteLocation createLocation = null;
     try {
       createLocation = rpcServer.getCreateLocation(src, locations);
-      return rpcClient.invokeSingle(createLocation, method,
+      HdfsFileStatus status = rpcClient.invokeSingle(createLocation, method,
           HdfsFileStatus.class);
+      status.setNsIdentify(createLocation.getNameserviceId());
+      return status;
     } catch (IOException ioe) {
       final List<RemoteLocation> newLocations = checkFaultTolerantRetry(
           method, src, ioe, createLocation, locations);
@@ -759,14 +764,47 @@ public boolean mkdirs(String src, FsPermission masked, boolean createParent)
     }
   }
 
+  /**
+   * Try to get a list of FederationNamespaceInfo for renewLease RPC.
+   */
+  private List<FederationNamespaceInfo> getRewLeaseNSs(String nsIdentifies)
+      throws IOException {
+    // return all namespaces
+    if (nsIdentifies == null || nsIdentifies.isEmpty()) {
+      return new ArrayList<>(namenodeResolver.getNamespaces());
+    }
+    String[] nsIdList = nsIdentifies.split(",");
+    List<FederationNamespaceInfo> result = new ArrayList<>();
+    for (String nsId : nsIdList) {
+      FederationNamespaceInfo namespaceInfo = nsNameSpaceInfoCache.get(nsId);
+      if (namespaceInfo == null) {
+        try {
+          rpcClient.getNamenodesForNameservice(nsId);
+        } catch (IOException ioe) {
+          // return all namespaces when parsing nsId failed.
+          return new ArrayList<>(namenodeResolver.getNamespaces());
+        }
+        namespaceInfo = new FederationNamespaceInfo("", "", nsId);
+        nsNameSpaceInfoCache.put(nsId, namespaceInfo);
+      }
+      result.add(namespaceInfo);
+    }
+    return result;
+  }
+
   @Override
-  public void renewLease(String clientName) throws IOException {
+  public void renewLease(String clientName, String nsIdentifies)
+      throws IOException {
     rpcServer.checkOperation(NameNode.OperationCategory.WRITE);
 
     RemoteMethod method = new RemoteMethod("renewLease",
-        new Class<?>[] {String.class}, clientName);
-    Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
-    rpcClient.invokeConcurrent(nss, method, false, false);
+        new Class<?>[] {String.class, String.class}, clientName, null);
+    List<FederationNamespaceInfo> nss = getRewLeaseNSs(nsIdentifies);
+    if (nss.size() == 1) {
+      rpcClient.invokeSingle(nss.get(0).getNameserviceId(), method);
+    } else {
+      rpcClient.invokeConcurrent(nss, method, false, false);
+    }
   }
 
   @Override

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

@@ -1508,7 +1508,7 @@ private void transferThreadLocalContext(
    * @return A prioritized list of NNs to use for communication.
    * @throws IOException If a NN cannot be located for the nameservice ID.
    */
-  private List<? extends FederationNamenodeContext> getNamenodesForNameservice(
+  public List<? extends FederationNamenodeContext> getNamenodesForNameservice(
       final String nsId) throws IOException {
 
     final List<? extends FederationNamenodeContext> namenodes =

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

@@ -980,8 +980,9 @@ public boolean mkdirs(String src, FsPermission masked, boolean createParent)
   }
 
   @Override // ClientProtocol
-  public void renewLease(String clientName) throws IOException {
-    clientProto.renewLease(clientName);
+  public void renewLease(String clientName, String nsIdentifies)
+      throws IOException {
+    clientProto.renewLease(clientName, nsIdentifies);
   }
 
   @Override // ClientProtocol

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/fairness/TestRouterHandlersFairness.java

@@ -194,7 +194,7 @@ private void invokeSequential(ClientProtocol routerProto) throws IOException {
 
   private void invokeConcurrent(ClientProtocol routerProto, String clientName)
       throws IOException {
-    routerProto.renewLease(clientName);
+    routerProto.renewLease(clientName, null);
   }
 
   private int getTotalRejectedPermits(RouterContext routerContext) {

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestRouterClientMetrics.java

@@ -156,7 +156,7 @@ public void testGetQuota() throws Exception {
 
   @Test
   public void testRenewLease() throws Exception {
-    router.getRpcServer().renewLease("test");
+    router.getRpcServer().renewLease("test", null);
     assertCounter("RenewLeaseOps", 2L, getMetrics(ROUTER_METRICS));
     assertCounter("ConcurrentRenewLeaseOps", 1L, getMetrics(ROUTER_METRICS));
   }

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

@@ -159,7 +159,7 @@ public void cleanup() throws IOException {
   public void testWithoutDisabling() throws IOException {
     // ns0 is slow and renewLease should take a long time
     long t0 = monotonicNow();
-    routerProtocol.renewLease("client0");
+    routerProtocol.renewLease("client0", null);
     long t = monotonicNow() - t0;
     assertTrue("It took too little: " + t + "ms",
         t > TimeUnit.SECONDS.toMillis(1));
@@ -178,7 +178,7 @@ public void testDisabling() throws Exception {
 
     // renewLease should be fast as we are skipping ns0
     long t0 = monotonicNow();
-    routerProtocol.renewLease("client0");
+    routerProtocol.renewLease("client0", null);
     long t = monotonicNow() - t0;
     assertTrue("It took too long: " + t + "ms",
         t < TimeUnit.SECONDS.toMillis(1));

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

@@ -215,7 +215,7 @@ public void run() {
             routerClient = new DFSClient(address, conf);
             String clientName = routerClient.getClientName();
             ClientProtocol routerProto = routerClient.getNamenode();
-            routerProto.renewLease(clientName);
+            routerProto.renewLease(clientName, null);
           } catch (RemoteException re) {
             IOException ioe = re.unwrapRemoteException();
             assertTrue("Wrong exception: " + ioe,
@@ -390,7 +390,7 @@ public void testAsyncCallerPoolMetrics() throws Exception {
               cluster.getRouterClientConf());
           String clientName = routerClient.getClientName();
           ClientProtocol routerProto = routerClient.getNamenode();
-          routerProto.renewLease(clientName);
+          routerProto.renewLease(clientName, null);
         } catch (Exception e) {
           fail("Client request failed: " + e);
         } finally {

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

@@ -153,7 +153,7 @@ public void testRetryWhenOneNameServiceDown() throws Exception {
 
     DFSClient client = nnContext1.getClient();
     // Renew lease for the DFS client, it will succeed.
-    routerProtocol.renewLease(client.getClientName());
+    routerProtocol.renewLease(client.getClientName(), null);
 
     // Verify the retry times, it will retry one time for ns0.
     FederationRPCMetrics rpcMetrics = routerContext.getRouter()