HDFS-16262. Async refresh of cached locations in DFSInputStream (apache#3527)

(cherry picked from commit 94b884a)

Author: bbeaudreault

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

@@ -69,6 +69,11 @@ public class ClientContext {
    */
   private final String name;
 
+  /**
+   * The client conf used to initialize context.
+   */
+  private final DfsClientConf dfsClientConf;
+
   /**
    * String representation of the configuration.
    */
@@ -130,6 +135,17 @@ public class ClientContext {
    */
   private volatile DeadNodeDetector deadNodeDetector = null;
 
+  /**
+   * The switch for the {@link LocatedBlocksRefresher}.
+   */
+  private final boolean locatedBlocksRefresherEnabled;
+
+  /**
+   * Periodically refresh the {@link org.apache.hadoop.hdfs.protocol.LocatedBlocks} backing
+   * registered {@link DFSInputStream}s, to take advantage of changes in block placement.
+   */
+  private volatile LocatedBlocksRefresher locatedBlocksRefresher = null;
+
   /**
    * Count the reference of ClientContext.
    */
@@ -146,6 +162,7 @@ private ClientContext(String name, DfsClientConf conf,
     final ShortCircuitConf scConf = conf.getShortCircuitConf();
 
     this.name = name;
+    this.dfsClientConf = conf;
     this.confString = scConf.confAsString();
     this.clientShortCircuitNum = conf.getClientShortCircuitNum();
     this.shortCircuitCache = new ShortCircuitCache[this.clientShortCircuitNum];
@@ -164,6 +181,7 @@ private ClientContext(String name, DfsClientConf conf,
     this.byteArrayManager = ByteArrayManager.newInstance(
         conf.getWriteByteArrayManagerConf());
     this.deadNodeDetectionEnabled = conf.isDeadNodeDetectionEnabled();
+    this.locatedBlocksRefresherEnabled = conf.isLocatedBlocksRefresherEnabled();
     initTopologyResolution(config);
   }
 
@@ -301,6 +319,21 @@ public DeadNodeDetector getDeadNodeDetector() {
     return deadNodeDetector;
   }
 
+  /**
+   * If true, LocatedBlocksRefresher will be periodically refreshing LocatedBlocks
+   * of registered DFSInputStreams.
+   */
+  public boolean isLocatedBlocksRefresherEnabled() {
+    return locatedBlocksRefresherEnabled;
+  }
+
+  /**
+   * Obtain LocatedBlocksRefresher of the current client.
+   */
+  public LocatedBlocksRefresher getLocatedBlocksRefresher() {
+    return locatedBlocksRefresher;
+  }
+
   /**
    * Increment the counter. Start the dead node detector thread if there is no
    * reference.
@@ -311,6 +344,10 @@ synchronized void reference() {
       deadNodeDetector = new DeadNodeDetector(name, configuration);
       deadNodeDetector.start();
     }
+    if (locatedBlocksRefresherEnabled && locatedBlocksRefresher == null) {
+      locatedBlocksRefresher = new LocatedBlocksRefresher(name, configuration, dfsClientConf);
+      locatedBlocksRefresher.start();
+    }
   }
 
   /**
@@ -324,5 +361,10 @@ synchronized void unreference() {
       deadNodeDetector.shutdown();
       deadNodeDetector = null;
     }
+
+    if (counter == 0 && locatedBlocksRefresherEnabled && locatedBlocksRefresher != null) {
+      locatedBlocksRefresher.shutdown();
+      locatedBlocksRefresher = null;
+    }
   }
 }

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

@@ -870,7 +870,7 @@ public void reportBadBlocks(LocatedBlock[] blocks) throws IOException {
   }
 
   public long getRefreshReadBlkLocationsInterval() {
-    return dfsClientConf.getRefreshReadBlockLocationsMS();
+    return dfsClientConf.getLocatedBlocksRefresherInterval();
   }
 
   public LocatedBlocks getLocatedBlocks(String src, long start)
@@ -3402,4 +3402,36 @@ private boolean isDeadNodeDetectionEnabled() {
   public DeadNodeDetector getDeadNodeDetector() {
     return clientContext.getDeadNodeDetector();
   }
+
+  /**
+   * Obtain LocatedBlocksRefresher of the current client.
+   */
+  public LocatedBlocksRefresher getLocatedBlockRefresher() {
+    return clientContext.getLocatedBlocksRefresher();
+  }
+
+  /**
+   * Adds the {@link DFSInputStream} to the {@link LocatedBlocksRefresher}, so that
+   * the underlying {@link LocatedBlocks} is periodically refreshed.
+   */
+  public void addLocatedBlocksRefresh(DFSInputStream dfsInputStream) {
+    if (isLocatedBlocksRefresherEnabled()) {
+      clientContext.getLocatedBlocksRefresher().addInputStream(dfsInputStream);
+    }
+  }
+
+  /**
+   * Removes the {@link DFSInputStream} from the {@link LocatedBlocksRefresher}, so that
+   * the underlying {@link LocatedBlocks} is no longer periodically refreshed.
+   * @param dfsInputStream
+   */
+  public void removeLocatedBlocksRefresh(DFSInputStream dfsInputStream) {
+    if (isLocatedBlocksRefresherEnabled()) {
+      clientContext.getLocatedBlocksRefresher().removeInputStream(dfsInputStream);
+    }
+  }
+
+  private boolean isLocatedBlocksRefresherEnabled() {
+    return clientContext.isLocatedBlocksRefresherEnabled();
+  }
 }