From 3ecbfdfdc279c065329af2d3cae3aa598a5e3398 Mon Sep 17 00:00:00 2001
From: Tsuyoshi Ozawa <ozawa@apache.org>
Date: Wed, 22 Jun 2016 17:13:50 -0700
Subject: [PATCH] MAPREDUCE-6721.
 mapreduce.reduce.shuffle.memory.limit.percent=0.0 should be legal to enforce
 shuffle to disk. (Gera Shegalov via ozawa)

This closes #102

(cherry picked from commit 79a7289165510072c46779251ebb010248cb0ce8)
---
 .../task/reduce/MergeManagerImpl.java          |  2 +-
 .../src/main/resources/mapred-default.xml      |  3 ++-
 .../task/reduce/TestMergeManager.java          | 18 ++++++++++++++++++
 3 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
index c99a330478a..1673ff80a7c 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
@@ -178,7 +178,7 @@ public class MergeManagerImpl<K, V> implements MergeManager<K, V> {
     final float singleShuffleMemoryLimitPercent =
         jobConf.getFloat(MRJobConfig.SHUFFLE_MEMORY_LIMIT_PERCENT,
             DEFAULT_SHUFFLE_MEMORY_LIMIT_PERCENT);
-    if (singleShuffleMemoryLimitPercent <= 0.0f
+    if (singleShuffleMemoryLimitPercent < 0.0f
         || singleShuffleMemoryLimitPercent > 1.0f) {
       throw new IllegalArgumentException("Invalid value for "
           + MRJobConfig.SHUFFLE_MEMORY_LIMIT_PERCENT + ": "
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
index 47b82608067..fab5b252310 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
@@ -443,7 +443,8 @@
   <name>mapreduce.reduce.shuffle.memory.limit.percent</name>
   <value>0.25</value>
   <description>Expert: Maximum percentage of the in-memory limit that a
-  single shuffle can consume</description>
+  single shuffle can consume. Range of valid values is [0.0, 1.0]. If the value
+  is 0.0 map outputs are shuffled directly to disk.</description>
 </property>
 
 <property>
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
index ef860afaf85..1c0d25b09a3 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMergeManager.java
@@ -41,6 +41,7 @@ import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.MROutputFiles;
 import org.apache.hadoop.mapred.MapOutputFile;
 import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
 import org.apache.hadoop.mapreduce.task.reduce.MergeManagerImpl.CompressAwarePath;
 import org.junit.Assert;
 import org.junit.Test;
@@ -289,4 +290,21 @@ public class TestMergeManager {
     assertTrue("Large in-memory reduce area unusable: " + maxInMemReduce,
         maxInMemReduce > Integer.MAX_VALUE);
   }
+
+  @Test
+  public void testZeroShuffleMemoryLimitPercent() throws Exception {
+    final JobConf jobConf = new JobConf();
+    jobConf.setFloat(MRJobConfig.SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f);
+    final MergeManager<Text, Text> mgr =
+        new MergeManagerImpl<>(null, jobConf, mock(LocalFileSystem.class),
+            null, null, null, null, null, null, null, null, null, null,
+            new MROutputFiles());
+    final long mapOutputSize = 10;
+    final int fetcher = 1;
+    final MapOutput<Text, Text> mapOutput = mgr.reserve(
+        TaskAttemptID.forName("attempt_0_1_m_1_1"),
+        mapOutputSize, fetcher);
+    assertEquals("Tiny map outputs should be shuffled to disk", "DISK",
+        mapOutput.getDescription());
+  }
 }