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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hdfs;
 
+import java.io.IOException;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.hadoop.classification.VisibleForTesting;
@@ -71,4 +72,6 @@ public void delayWhenRenewLeaseTimeout() {}
   public void onCreateBlockReader(LocatedBlock block, int chunkIndex, long offset, long length) {}
 
   public void failCreateBlockReader() throws InvalidBlockTokenException {}
+  
+  public void failCreateBlockReader(int curAttempt) 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 readDNMaxAttempts) throws IOException {
     BlockReader reader = null;
     final ReaderRetryPolicy retry = new ReaderRetryPolicy();
     DFSInputStream.DNAddrPair dnInfo =
         new DFSInputStream.DNAddrPair(null, null, null, null);
-
+    int curAttempts = 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(curAttempts);
         reader = getBlockReader(block, offsetInBlock,
             readTo - offsetInBlock,
             dnInfo.addr, dnInfo.storageType, dnInfo.info);
@@ -273,9 +274,14 @@ boolean createBlockReader(LocatedBlock block, long offsetInBlock,
           fetchBlockAt(block.getStartOffset());
           retry.refetchToken();
         } else {
+          if (curAttempts++ < readDNMaxAttempts) {
+            DFSClient.LOG.info("Try to reconnect to {} for block {}, retry {} time.", dnInfo.addr,
+                block.getBlock(), curAttempts);
+            continue;
+          }
           //TODO: handles connection issues
-          DFSClient.LOG.warn("Failed to connect to " + dnInfo.addr + " for " +
-              "block" + block.getBlock(), e);
+          DFSClient.LOG.warn("Failed to connect to {} for block {} after {} time.", dnInfo.addr,
+              block.getBlock(), curAttempts, 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 readDNMaxAttempts;
 
   protected ECChunk[] decodeInputs;
 
@@ -138,6 +139,8 @@ void skip() {
     this.corruptedBlocks = corruptedBlocks;
     this.decoder = decoder;
     this.dfsStripedInputStream = dfsStripedInputStream;
+    this.readDNMaxAttempts = dfsStripedInputStream.getDFSClient()
+        .getConf().getStripedReadDnMaxAttempts();
 
     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, readDNMaxAttempts)) {
         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 stripedReadDnMaxAttempts;
 
   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.");
+    stripedReadDnMaxAttempts =
+        conf.getInt(
+            HdfsClientConfigKeys.StripedRead.DATANODE_MAX_RETRY_COUNT,
+            HdfsClientConfigKeys.StripedRead.DATANODE_MAX_RETRY_COUNT_DEFAULT);
+    Preconditions.checkArgument(stripedReadDnMaxAttempts >= 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 stripedReadDnMaxAttempts
+   */
+  public int getStripedReadDnMaxAttempts() {
+    return stripedReadDnMaxAttempts;
+  }
+
   /**
    * @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>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedInputStream.java

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hdfs;
 
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.HadoopIllegalArgumentException;
@@ -50,6 +52,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Random;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
@@ -735,4 +738,50 @@ public void onCreateBlockReader(LocatedBlock block, int chunkIndex,
     assertEquals(rangesExpected, ranges);
   }
 
+  @Test
+  public void testStatefulReadAfterLongTimeIdle() throws Exception {
+    HdfsConfiguration hdfsConf = new HdfsConfiguration();
+    // Set retry count to 0 to reproduce the issue.
+    hdfsConf.setInt(HdfsClientConfigKeys.StripedRead.DATANODE_MAX_RETRY_COUNT, 1);
+    String testBaseDir = "/testECRead";
+    String testfileName = "testfile";
+    DFSClientFaultInjector old = DFSClientFaultInjector.get();
+    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(hdfsConf)
+        .numDataNodes(9).build()) {
+      cluster.waitActive();
+      final DistributedFileSystem dfs = cluster.getFileSystem();
+      Path dir = new Path(testBaseDir);
+      assertTrue(dfs.mkdirs(dir));
+      dfs.enableErasureCodingPolicy("RS-6-3-1024k");
+      dfs.setErasureCodingPolicy(dir, "RS-6-3-1024k");
+      assertEquals("RS-6-3-1024k", dfs.getErasureCodingPolicy(dir).getName());
+
+      int writeBufSize = 30 * 1024 * 1024 + 1;
+      byte[] writeBuf = new byte[writeBufSize];
+      try (FSDataOutputStream fsdos = dfs.create(
+          new Path(testBaseDir + Path.SEPARATOR + testfileName))) {
+        Random random = new Random();
+        random.nextBytes(writeBuf);
+        fsdos.write(writeBuf, 0, writeBuf.length);
+        Thread.sleep(2000);
+      }
+
+      DFSClientFaultInjector.set(new DFSClientFaultInjector() {
+        public void failCreateBlockReader(int curAttempt) throws IOException {
+          if (0 == curAttempt) {
+            throw new IOException("Mock IOException when first create blockread.");
+          }
+        }
+      });
+
+      byte[] readBuf = new byte[6 * 1024 * 1024];
+      try (FSDataInputStream fsdis = dfs.open(
+          new Path(testBaseDir + Path.SEPARATOR + testfileName))) {
+        while ((fsdis.read(readBuf)) > 0) {}
+      }
+      assertTrue(dfs.delete(new Path(testBaseDir + Path.SEPARATOR + testfileName), true));
+    } finally {
+      DFSClientFaultInjector.set(old);
+    }
+  }
 }