HDFS-17365. EC: Add extra redunency configuration in checkStreamerFailures to prevent data loss. (#6517) Contributed by hfutatzhanghb.

Reviewed-by: Takanobu Asanuma <tasanuma@apache.org>
Signed-off-by: Shuyan Zhang <zhangshuyan@apache.org>

Author: hfutatzhanghb

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

@@ -73,6 +73,9 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.Write.ECRedundancy.DFS_CLIENT_EC_WRITE_FAILED_BLOCKS_TOLERATED;
+import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.Write.ECRedundancy.DFS_CLIENT_EC_WRITE_FAILED_BLOCKS_TOLERATED_DEFAILT;
+
 /**
  * This class supports writing files in striped layout and erasure coded format.
  * Each stripe contains a sequence of cells.
@@ -283,6 +286,7 @@ private void flipDataBuffers() {
   private CompletionService<Void> flushAllExecutorCompletionService;
   private int blockGroupIndex;
   private long datanodeRestartTimeout;
+  private final int failedBlocksTolerated;
 
   /** Construct a new output stream for creating a file. */
   DFSStripedOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
@@ -322,6 +326,15 @@ private void flipDataBuffers() {
     currentPackets = new DFSPacket[streamers.size()];
     datanodeRestartTimeout = dfsClient.getConf().getDatanodeRestartTimeout();
     setCurrentStreamer(0);
+
+    int failedBlocksToleratedTmp = dfsClient.getConfiguration().getInt(
+        DFS_CLIENT_EC_WRITE_FAILED_BLOCKS_TOLERATED,
+        DFS_CLIENT_EC_WRITE_FAILED_BLOCKS_TOLERATED_DEFAILT);
+    if (failedBlocksToleratedTmp < 0) {
+      failedBlocksToleratedTmp = ecPolicy.getNumParityUnits();
+    }
+    failedBlocksTolerated = Math.min(failedBlocksToleratedTmp,
+        ecPolicy.getNumParityUnits());
   }
 
   /** Construct a new output stream for appending to a file. */
@@ -402,11 +415,11 @@ private Set<StripedDataStreamer> checkStreamers() throws IOException {
       LOG.debug("original failed streamers: {}", failedStreamers);
       LOG.debug("newly failed streamers: {}", newFailed);
     }
-    if (failCount > (numAllBlocks - numDataBlocks)) {
+    if (failCount > failedBlocksTolerated) {
       closeAllStreamers();
       throw new IOException("Failed: the number of failed blocks = "
-          + failCount + " > the number of parity blocks = "
-          + (numAllBlocks - numDataBlocks));
+          + failCount + " > the number of failed blocks tolerated = "
+          + failedBlocksTolerated);
     }
     return newFailed;
   }
@@ -687,7 +700,7 @@ private void checkStreamerFailures(boolean isNeedFlushAllPackets)
       // 2) create new block outputstream
       newFailed = waitCreatingStreamers(healthySet);
       if (newFailed.size() + failedStreamers.size() >
-          numAllBlocks - numDataBlocks) {
+          failedBlocksTolerated) {
         // The write has failed, Close all the streamers.
         closeAllStreamers();
         throw new IOException(

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

@@ -427,6 +427,13 @@ interface ByteArrayManager {
           PREFIX + "count-reset-time-period-ms";
       long    COUNT_RESET_TIME_PERIOD_MS_DEFAULT = 10 * MS_PER_SECOND;
     }
+
+    @SuppressWarnings("checkstyle:InterfaceIsType")
+    interface ECRedundancy {
+      String DFS_CLIENT_EC_WRITE_FAILED_BLOCKS_TOLERATED =
+          "dfs.client.ec.write.failed.blocks.tolerated";
+      int DFS_CLIENT_EC_WRITE_FAILED_BLOCKS_TOLERATED_DEFAILT = -1;
+    }
   }
 
   /** dfs.client.block.write configuration properties */

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

@@ -3923,6 +3923,19 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.client.ec.write.failed.blocks.tolerated</name>
+  <value>-1</value>
+  <description>
+    Provide extra tolerated failed streamer for ec policy to prevent
+    the potential data loss. For example, if we use RS-6-3-1024K ec policy.
+    We can write successfully when there are 3 failure streamers. But if one of the six
+    replicas lost during reconstruction, we may lose the data forever.
+    It should better configured between [0, numParityBlocks], the default value is -1 which
+    means the parity block number of the specified ec policy we are using.
+  </description>
+</property>
+
 <property>
   <name>dfs.namenode.quota.init-threads</name>
   <value>12</value>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/tools/TestHdfsConfigFields.java

@@ -47,7 +47,8 @@ public void initializeMemberVariables() {
         HdfsClientConfigKeys.Read.class, HdfsClientConfigKeys.HedgedRead.class,
         HdfsClientConfigKeys.ShortCircuit.class,
         HdfsClientConfigKeys.Retry.class, HdfsClientConfigKeys.Mmap.class,
-        HdfsClientConfigKeys.BlockWrite.ReplaceDatanodeOnFailure.class };
+        HdfsClientConfigKeys.BlockWrite.ReplaceDatanodeOnFailure.class,
+        HdfsClientConfigKeys.Write.ECRedundancy.class};
 
     // Set error modes
     errorIfMissingConfigProps = true;