HDFS-17801. EC: Reading support retryCurrentNode to avoid transient errors cause application level failures.

Author: hfutatzhanghb

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

@@ -206,6 +206,9 @@ protected void closeCurrentBlockReaders() {
       return;
     }
     for (int i = 0; i < groupSize; i++) {
+      if (blockReaders[i] != null) {
+        blockReaders[i].setRetryCurrentReader(false);
+      }
       closeReader(blockReaders[i]);
       blockReaders[i] = null;
     }
@@ -220,7 +223,9 @@ protected void closeReader(BlockReaderInfo readerInfo) {
         } catch (Throwable ignored) {
         }
       }
-      readerInfo.skip();
+      if (!readerInfo.isRetryCurrentReader()) {
+        readerInfo.skip();
+      }
     }
   }
 
@@ -517,6 +522,9 @@ protected void fetchBlockByteRange(LocatedBlock block, long start,
       buf.position(buf.position() + (int)(end - start + 1));
     } finally {
       for (BlockReaderInfo preaderInfo : preaderInfos) {
+        if (preaderInfo != null) {
+          preaderInfo.setRetryCurrentReader(false);
+        }
         closeReader(preaderInfo);
       }
     }

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

@@ -38,7 +38,9 @@
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CompletionService;
 import java.util.concurrent.ExecutorCompletionService;
@@ -90,6 +92,13 @@ static class BlockReaderInfo {
      * using it for the next stripe.
      */
     boolean shouldSkip = false;
+    /**
+     * We use this field to indicate whether we should retry the current reader before
+     * we mark current reader skipped. possibly retry the same node so that transient errors don't
+     * result in application level failures (e.g. Datanode could have closed the connection
+     * because the client is idle for too long).
+     */
+    boolean retryCurrentReader = true;
 
     BlockReaderInfo(BlockReader reader, DatanodeInfo dn, long offset) {
       this.reader = reader;
@@ -104,6 +113,14 @@ void setOffset(long offset) {
     void skip() {
       this.shouldSkip = true;
     }
+
+    public boolean isRetryCurrentReader() {
+      return retryCurrentReader;
+    }
+
+    public void setRetryCurrentReader(boolean retryCurrentReader) {
+      this.retryCurrentReader = retryCurrentReader;
+    }
   }
 
   private final Map<Future<BlockReadStats>, Integer> futures =
@@ -174,11 +191,26 @@ void updateState4SuccessRead(StripingChunkReadResult result) {
 
   private void checkMissingBlocks() throws IOException {
     if (alignedStripe.missingChunksNum > parityBlkNum) {
-      clearFutures();
-      throw new IOException(alignedStripe.missingChunksNum
-          + " missing blocks, the stripe is: " + alignedStripe
-          + "; locatedBlocks is: " + dfsStripedInputStream.getLocatedBlocks());
+      if (countOfNullReaderInfos(readerInfos) < parityBlkNum) {
+        clearFutures();
+        throw new IOException(alignedStripe.missingChunksNum
+            + " missing blocks, the stripe is: " + alignedStripe
+            + "; locatedBlocks is: " + dfsStripedInputStream.getLocatedBlocks());
+      }
+    }
+  }
+
+  private int countOfNullReaderInfos(BlockReaderInfo[] readerInfos) {
+    if (readerInfos == null) {
+      return 0;
     }
+    int count = 0;
+    for (int i = 0; i < readerInfos.length; i++) {
+      if (readerInfos[i] == null) {
+        count++;
+      }
+    }
+    return count;
   }
 
   /**
@@ -187,6 +219,23 @@ private void checkMissingBlocks() throws IOException {
    */
   private void readDataForDecoding() throws IOException {
     prepareDecodeInputs();
+
+    if (alignedStripe.missingChunksNum > parityBlkNum) {
+      Set<Integer> recoveredIndexes = new HashSet<>();
+      if (countOfNullReaderInfos(readerInfos) >= parityBlkNum) {
+        for (int index = 0; index < dataBlkNum + parityBlkNum; index++) {
+          if (readerInfos[index] == null) {
+            alignedStripe.chunks[index].state = StripingChunk.REQUESTED;
+            recoveredIndexes.add(index);
+          }
+        }
+      }
+
+      for (int recoveredIndex : recoveredIndexes) {
+        alignedStripe.missingChunksNum--;
+      }
+    }
+    
     for (int i = 0; i < dataBlkNum; i++) {
       Preconditions.checkNotNull(alignedStripe.chunks[i]);
       if (alignedStripe.chunks[i].state == StripingChunk.REQUESTED) {
@@ -332,7 +381,7 @@ boolean readChunk(final LocatedBlock block, int chunkIndex)
   }
 
   /**
-   * read the whole stripe. do decoding if necessary
+   * Read the whole stripe. do decoding if necessary.
    */
   void readStripe() throws IOException {
     try {
@@ -349,7 +398,7 @@ void readStripe() throws IOException {
       if (alignedStripe.missingChunksNum > 0) {
         checkMissingBlocks();
         readDataForDecoding();
-        // read parity chunks
+        // Read parity chunks.
         readParityChunks(alignedStripe.missingChunksNum);
       }
     } catch (IOException e) {
@@ -359,7 +408,7 @@ void readStripe() throws IOException {
     // TODO: for a full stripe we can start reading (dataBlkNum + 1) chunks
 
     // Input buffers for potential decode operation, which remains null until
-    // first read failure
+    // first read failure.
     while (!futures.isEmpty()) {
       try {
         long beginReadMS = Time.monotonicNow();
@@ -384,8 +433,12 @@ void readStripe() throws IOException {
           }
         } else {
           returnedChunk.state = StripingChunk.MISSING;
-          // close the corresponding reader
+          // Close the corresponding reader.
           dfsStripedInputStream.closeReader(readerInfos[r.index]);
+          if (readerInfos[r.index].isRetryCurrentReader()) {
+            readerInfos[r.index].setRetryCurrentReader(false);
+            readerInfos[r.index] = null;
+          }
 
           final int missing = alignedStripe.missingChunksNum;
           alignedStripe.missingChunksNum++;
@@ -399,7 +452,7 @@ void readStripe() throws IOException {
         DFSClient.LOG.error(err, ie);
         dfsStripedInputStream.close();
         clearFutures();
-        // Don't decode if read interrupted
+        // Don't decode if read interrupted.
         throw new InterruptedIOException(err);
       }
     }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hdfs;
 
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.HadoopIllegalArgumentException;
@@ -50,6 +51,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 +737,87 @@ public void onCreateBlockReader(LocatedBlock block, int chunkIndex,
     assertEquals(rangesExpected, ranges);
   }
 
+  @Test
+  public void testStatefulReadAfterLongTimeIdle() throws Exception {
+    HdfsConfiguration hdfsConf = new HdfsConfiguration();
+    hdfsConf.setInt("dfs.datanode.socket.write.timeout", 10000);
+    hdfsConf.setInt("dfs.client.socket-timeout", 10000);
+    String testBaseDir = "/testECRead";
+    String testfileName = "testfile";
+    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(hdfsConf)
+        .numDataNodes(9).build()) {
+      cluster.waitActive();
+      final DistributedFileSystem dfs = cluster.getFileSystem();
+      Path dir = new Path(testBaseDir);
+      dfs.mkdirs(dir);
+      dfs.enableErasureCodingPolicy("RS-6-3-1024k");
+      dfs.setErasureCodingPolicy(dir, "RS-6-3-1024k");
+
+      
+      // Write 30MB + 1 data.
+      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);
+      }
+
+      byte[] readBuf = new byte[6 * 1024 * 1024];
+      try (FSDataInputStream fsdis = dfs.open(
+          new Path(testBaseDir + Path.SEPARATOR + testfileName))) {
+        fsdis.read(readBuf);
+        Thread.sleep(21 * 1000);
+        while ((fsdis.read(readBuf)) > 0) {
+          Thread.sleep(21 * 1000);
+        }
+      }
+      assertTrue(dfs.delete(new Path(testBaseDir + Path.SEPARATOR + testfileName), true));
+    }
+  }
+
+  @Test
+  public void testPReadAfterLongTimeIdle() throws Exception {
+    HdfsConfiguration hdfsConf = new HdfsConfiguration();
+    hdfsConf.setInt("dfs.datanode.socket.write.timeout", 10000);
+    hdfsConf.setInt("dfs.client.socket-timeout", 10000);
+    String testBaseDir = "/testECRead";
+    String testfileName = "testfile";
+    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(hdfsConf)
+        .numDataNodes(9).build()) {
+      cluster.waitActive();
+      final DistributedFileSystem dfs = cluster.getFileSystem();
+      Path dir = new Path(testBaseDir);
+      dfs.mkdirs(dir);
+      dfs.enableErasureCodingPolicy("RS-6-3-1024k");
+      dfs.setErasureCodingPolicy(dir, "RS-6-3-1024k");
+
+      // Write 30MB + 1 data.
+      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);
+      }
+
+      byte[] readBuf = new byte[6 * 1024 * 1024];
+      try (FSDataInputStream fsdis = dfs.open(
+          new Path(testBaseDir + Path.SEPARATOR + testfileName))) {
+        int curPos = 0;
+        int readLen = fsdis.read(curPos, readBuf, 0, readBuf.length);
+        curPos += readLen;
+        Thread.sleep(21 * 1000);
+        while ((readLen = fsdis.read(curPos, readBuf, 0, readBuf.length)) > 0) {
+          curPos += readLen;
+          Thread.sleep(21 * 1000);
+        }
+      }
+      assertTrue(dfs.delete(new Path(testBaseDir + Path.SEPARATOR + testfileName), true));
+    }
+  }
 }