HDFS-17801. EC supports createBlockReader retry when IOException occurs occasionally.

Author: hfutatzhanghb

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

@@ -72,6 +72,8 @@ public void delayWhenRenewLeaseTimeout() {}
   public void onCreateBlockReader(LocatedBlock block, int chunkIndex, long offset, long length) {}
 
   public void failCreateBlockReader() throws InvalidBlockTokenException {}
+  
+  public void failWhenReadWithStrategy(boolean isRetryRead) throws IOException {}
 
-  public void failWhenReadWithStrategy(boolean isRetryRead) throws IOException {};
+  public void failCreateBlockReader(int curRetryCounts) throws IOException {}
 }

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

@@ -234,12 +234,12 @@ private long getOffsetInBlockGroup(long pos) {
 
   boolean createBlockReader(LocatedBlock block, long offsetInBlock,
       LocatedBlock[] targetBlocks, BlockReaderInfo[] readerInfos,
-      int chunkIndex, long readTo) throws IOException {
+      int chunkIndex, long readTo, int readDNMaxRetryCounts) throws IOException {
     BlockReader reader = null;
     final ReaderRetryPolicy retry = new ReaderRetryPolicy();
     DFSInputStream.DNAddrPair dnInfo =
         new DFSInputStream.DNAddrPair(null, null, null, null);
-
+    int curRetryCounts = 0;
     while (true) {
       try {
         // the cached block location might have been re-fetched, so always
@@ -255,6 +255,7 @@ boolean createBlockReader(LocatedBlock block, long offsetInBlock,
         if (readTo < 0 || readTo > block.getBlockSize()) {
           readTo = block.getBlockSize();
         }
+        DFSClientFaultInjector.get().failCreateBlockReader(curRetryCounts);
         reader = getBlockReader(block, offsetInBlock,
             readTo - offsetInBlock,
             dnInfo.addr, dnInfo.storageType, dnInfo.info);
@@ -273,10 +274,17 @@ boolean createBlockReader(LocatedBlock block, long offsetInBlock,
           fetchBlockAt(block.getStartOffset());
           retry.refetchToken();
         } else {
+          if (curRetryCounts++ < readDNMaxRetryCounts) {
+            DFSClient.LOG.info("Try to reconnect to {} for block {} for {} time.", dnInfo.addr,
+                block.getBlock(), curRetryCounts);
+            // Re-fetch the block in case the block has been moved.
+            fetchBlockAt(block.getStartOffset());
+            continue;
+          }
           //TODO: handles connection issues
-          DFSClient.LOG.warn("Failed to connect to " + dnInfo.addr + " for " +
-              "block" + block.getBlock(), e);
-          // re-fetch the block in case the block has been moved
+          DFSClient.LOG.warn("Failed to connect to {} for block {} after retrying {} time.",
+              dnInfo.addr, block.getBlock(), curRetryCounts, e);
+          // Re-fetch the block in case the block has been moved
           fetchBlockAt(block.getStartOffset());
           addToLocalDeadNodes(dnInfo.info);
         }

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

@@ -120,6 +120,7 @@ void skip() {
   protected final RawErasureDecoder decoder;
   protected final DFSStripedInputStream dfsStripedInputStream;
   private long readTo = -1;
+  private final int readDNMaxRetryCounts;
 
   protected ECChunk[] decodeInputs;
 
@@ -138,6 +139,8 @@ void skip() {
     this.corruptedBlocks = corruptedBlocks;
     this.decoder = decoder;
     this.dfsStripedInputStream = dfsStripedInputStream;
+    this.readDNMaxRetryCounts = dfsStripedInputStream.getDFSClient()
+        .getConf().getStripedReadDnMaxRetryCounts();
 
     service = new ExecutorCompletionService<>(
             dfsStripedInputStream.getStripedReadsThreadPool());
@@ -309,7 +312,7 @@ boolean readChunk(final LocatedBlock block, int chunkIndex)
     if (readerInfos[chunkIndex] == null) {
       if (!dfsStripedInputStream.createBlockReader(block,
           alignedStripe.getOffsetInBlock(), targetBlocks,
-          readerInfos, chunkIndex, readTo)) {
+          readerInfos, chunkIndex, readTo, readDNMaxRetryCounts)) {
         chunk.state = StripingChunk.MISSING;
         return false;
       }

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

@@ -532,6 +532,14 @@ interface StripedRead {
      * span 6 DNs, so this default value accommodates 3 read streams
      */
     int     THREADPOOL_SIZE_DEFAULT = 18;
+
+    /**
+     * The max retry counts of reconnecting to DN
+     * during the striped read process, Default is 1.
+     */
+    String DATANODE_MAX_RETRY_COUNT = PREFIX +
+        "datanode.max.retry.count";
+    int DATANODE_MAX_RETRY_COUNT_DEFAULT = 1;
   }
 
   /** dfs.http.client configuration properties */

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

@@ -167,6 +167,7 @@ public class DfsClientConf {
   private final boolean deadNodeDetectionEnabled;
   private final long leaseHardLimitPeriod;
   private final boolean recoverLeaseOnCloseException;
+  private final int stripedReadDnMaxRetryCounts;
 
   public DfsClientConf(Configuration conf) {
     // The hdfsTimeout is currently the same as the ipc timeout
@@ -296,6 +297,13 @@ public DfsClientConf(Configuration conf) {
     Preconditions.checkArgument(stripedReadThreadpoolSize > 0, "The value of " +
         HdfsClientConfigKeys.StripedRead.THREADPOOL_SIZE_KEY +
         " must be greater than 0.");
+    stripedReadDnMaxRetryCounts =
+        conf.getInt(
+            HdfsClientConfigKeys.StripedRead.DATANODE_MAX_RETRY_COUNT,
+            HdfsClientConfigKeys.StripedRead.DATANODE_MAX_RETRY_COUNT_DEFAULT);
+    Preconditions.checkArgument(stripedReadDnMaxRetryCounts >= 0, "The value of " +
+        HdfsClientConfigKeys.StripedRead.DATANODE_MAX_RETRY_COUNT +
+        " must be greater than or equal to 0.");
     replicaAccessorBuilderClasses = loadReplicaAccessorBuilderClasses(conf);
 
     leaseHardLimitPeriod =
@@ -710,6 +718,13 @@ public long getleaseHardLimitPeriod() {
     return leaseHardLimitPeriod;
   }
 
+  /**
+   * @return the stripedReadDnMaxRetryCounts
+   */
+  public int getStripedReadDnMaxRetryCounts() {
+    return stripedReadDnMaxRetryCounts;
+  }
+
   /**
    * @return the readUseCachePriority
    */

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -4625,6 +4625,15 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.client.read.striped.datanode.max.retry.count</name>
+  <value>1</value>
+  <description>
+    The max retry counts of reconnecting to DN during
+    the striped read process, Default is 1.
+  </description>
+</property>
+
 <property>
   <name>dfs.client.replica.accessor.builder.classes</name>
   <value></value>