HDFS-17577.Add Support for CreateFlag.NO_LOCAL_WRITE in File Creation to Manage Disk Space and Network Load in Labeled YARN Nodes

Author: liangyu-1

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

@@ -503,10 +503,17 @@ public FSDataOutputStream next(final FileSystem fs, final Path p)
   public FSDataOutputStream create(Path f, FsPermission permission,
       boolean overwrite, int bufferSize, short replication, long blockSize,
       Progressable progress) throws IOException {
+    boolean noLocalWrite =dfs.getConf().getNoLocalWrite();
+    EnumSet<CreateFlag> flag = EnumSet.of(CreateFlag.CREATE);
+    if(overwrite){
+      flag.add(CreateFlag.OVERWRITE);
+    }
+    if(noLocalWrite){
+      flag.add(CreateFlag.NO_LOCAL_WRITE);
+    }
     return this.create(f, permission,
-        overwrite ? EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
-            : EnumSet.of(CreateFlag.CREATE), bufferSize, replication,
-        blockSize, progress, null);
+            flag, bufferSize, replication,
+            blockSize, progress, null);
   }
 
   /**

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

@@ -167,6 +167,8 @@ public interface HdfsClientConfigKeys {
       "dfs.client.block.reader.remote.buffer.size";
   int DFS_CLIENT_BLOCK_READER_REMOTE_BUFFER_SIZE_DEFAULT = 512;
 
+  String DFS_CLIENT_NO_LOCAL_WRITE = "dfs.client.write.no_local_write";
+  boolean DFS_CLIENT_NO_LOCAL_WRITE_DEFAULT = false;
   String DFS_CLIENT_DEAD_NODE_DETECTION_ENABLED_KEY =
           "dfs.client.deadnode.detection.enabled";
   boolean DFS_CLIENT_DEAD_NODE_DETECTION_ENABLED_DEFAULT = false;

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

@@ -152,6 +152,7 @@ public class DfsClientConf {
 
   private final ShortCircuitConf shortCircuitConf;
   private final int clientShortCircuitNum;
+  private final boolean noLocalWrite;
 
   private final long hedgedReadThresholdMillis;
   private final int hedgedReadThreadpoolSize;
@@ -311,6 +312,8 @@ public DfsClientConf(Configuration conf) {
     Preconditions.checkArgument(clientShortCircuitNum <= 5,
             HdfsClientConfigKeys.DFS_CLIENT_SHORT_CIRCUIT_NUM +
                     "can't be more then 5.");
+    noLocalWrite = conf.getBoolean(HdfsClientConfigKeys.DFS_CLIENT_NO_LOCAL_WRITE,
+            HdfsClientConfigKeys.DFS_CLIENT_NO_LOCAL_WRITE_DEFAULT);
     maxPipelineRecoveryRetries = conf.getInt(
         HdfsClientConfigKeys.DFS_CLIENT_PIPELINE_RECOVERY_MAX_RETRIES,
         HdfsClientConfigKeys.DFS_CLIENT_PIPELINE_RECOVERY_MAX_RETRIES_DEFAULT
@@ -745,6 +748,12 @@ public int getMaxPipelineRecoveryRetries() {
     return maxPipelineRecoveryRetries;
   }
 
+  /**
+   *@return the noLocalWrite
+   */
+  public boolean getNoLocalWrite() {
+    return noLocalWrite;
+  }
   /**
    * Configuration for short-circuit reads.
    */