MAPREDUCE-834. Enables memory management on tasktrackers when old mem…

…ory management parameters are used in configuration. Contributed by Sreekanth Ramakrishnan.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20@806422 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -221,6 +221,10 @@ Release 0.20.1 - Unreleased
     MAPREDUCE-745. Fixes a testcase problem to do with generation of JobTracker
     IDs. (Amar Kamat via ddas)
 
+    MAPREDUCE-834. Enables memory management on tasktrackers when old
+    memory management parameters are used in configuration.
+    (Sreekanth Ramakrishnan via yhemanth)
+
 Release 0.20.0 - 2009-04-15
 
   INCOMPATIBLE CHANGES

src/mapred/org/apache/hadoop/mapred/TaskTracker.java

@@ -3081,7 +3081,26 @@ private void initializeMemoryManagement() {
         maxCurrentMapTasks * mapSlotMemorySizeOnTT + maxCurrentReduceTasks
             * reduceSlotSizeMemoryOnTT;
     if (totalMemoryAllottedForTasks < 0) {
-      totalMemoryAllottedForTasks = JobConf.DISABLED_MEMORY_LIMIT;
+      //adding check for the old keys which might be used by the administrator
+      //while configuration of the memory monitoring on TT
+      long memoryAllotedForSlot = fConf.normalizeMemoryConfigValue(
+          fConf.getLong(JobConf.MAPRED_TASK_DEFAULT_MAXVMEM_PROPERTY, 
+              JobConf.DISABLED_MEMORY_LIMIT));
+      long limitVmPerTask = fConf.normalizeMemoryConfigValue(
+          fConf.getLong(JobConf.UPPER_LIMIT_ON_TASK_VMEM_PROPERTY, 
+              JobConf.DISABLED_MEMORY_LIMIT));
+      if(memoryAllotedForSlot == JobConf.DISABLED_MEMORY_LIMIT) {
+        totalMemoryAllottedForTasks = JobConf.DISABLED_MEMORY_LIMIT; 
+      } else {
+        if(memoryAllotedForSlot > limitVmPerTask) {
+          LOG.info("DefaultMaxVmPerTask is mis-configured. " +
+          		"It shouldn't be greater than task limits");
+          totalMemoryAllottedForTasks = JobConf.DISABLED_MEMORY_LIMIT;
+        } else {
+          totalMemoryAllottedForTasks = (maxCurrentMapTasks + 
+              maxCurrentReduceTasks) *  (memoryAllotedForSlot/(1024 * 1024));
+        }
+      }
     }
     if (totalMemoryAllottedForTasks > totalPhysicalMemoryOnTT) {
       LOG.info("totalMemoryAllottedForTasks > totalPhysicalMemoryOnTT."

src/test/org/apache/hadoop/mapred/TestTaskTrackerMemoryManager.java

@@ -200,16 +200,8 @@ public void testTasksBeyondLimits()
       return;
     }
 
-    long PER_TASK_LIMIT = 1L; // Low enough to kill off sleepJob tasks.
-
-    Pattern taskOverLimitPattern =
-        Pattern.compile(String.format(taskOverLimitPatternString, String
-            .valueOf(PER_TASK_LIMIT*1024*1024L)));
-    Matcher mat = null;
-
     // Start cluster with proper configuration.
     JobConf fConf = new JobConf();
-
     // very small value, so that no task escapes to successful completion.
     fConf.set("mapred.tasktracker.taskmemorymanager.monitoring-interval",
         String.valueOf(300));
@@ -219,6 +211,51 @@ public void testTasksBeyondLimits()
         JobTracker.MAPRED_CLUSTER_REDUCE_MEMORY_MB_PROPERTY,
         2 * 1024);
     startCluster(fConf);
+    runJobExceedingMemoryLimit();
+  }
+
+  /**
+   * Runs tests with tasks beyond limit and using old configuration values for
+   * the TaskTracker.
+   * 
+   * @throws Exception
+   */
+
+  public void testTaskMemoryMonitoringWithDeprecatedConfiguration () 
+    throws Exception {
+
+    // Run the test only if memory management is enabled
+    if (!isProcfsBasedTreeAvailable()) {
+      return;
+    }
+    // Start cluster with proper configuration.
+    JobConf fConf = new JobConf();
+    // very small value, so that no task escapes to successful completion.
+    fConf.set("mapred.tasktracker.taskmemorymanager.monitoring-interval",
+        String.valueOf(300));
+    //set old values, max vm property per task and upper limit on the tasks
+    //vm
+    //setting the default maximum vmem property to 2 GB
+    fConf.setLong(JobConf.MAPRED_TASK_DEFAULT_MAXVMEM_PROPERTY,
+        (2L * 1024L * 1024L * 1024L));
+    fConf.setLong(JobConf.UPPER_LIMIT_ON_TASK_VMEM_PROPERTY, 
+        (3L * 1024L * 1024L * 1024L));
+    startCluster(fConf);
+    runJobExceedingMemoryLimit();
+  }
+
+  /**
+   * Runs a job which should fail the when run by the memory monitor.
+   * 
+   * @throws IOException
+   */
+  private void runJobExceedingMemoryLimit() throws IOException {
+    long PER_TASK_LIMIT = 1L; // Low enough to kill off sleepJob tasks.
+
+    Pattern taskOverLimitPattern =
+        Pattern.compile(String.format(taskOverLimitPatternString, String
+            .valueOf(PER_TASK_LIMIT*1024*1024L)));
+    Matcher mat = null;
 
     // Set up job.
     JobConf conf = new JobConf(miniMRCluster.createJobConf());