diff --git a/hadoop-mapreduce-project/CHANGES.txt b/hadoop-mapreduce-project/CHANGES.txt
index e60dd00ad0..2291e58d76 100644
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@@ -255,6 +255,10 @@ Release 2.4.0 - UNRELEASED
     MAPREDUCE-5570. Map task attempt with fetch failure has incorrect attempt
     finish time (Rushabh S Shah via jlowe)
 
+    MAPREDUCE-5806. Fixed a bug in MRAppMaster so as to enable users to properly
+    override HADOOP_ROOT_LOGGER or HADOOP_CLIENT_OPTS. (Varun Vasudev via
+    vinodkv)
+
 Release 2.3.1 - UNRELEASED
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/MapReduceChildJVM.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/MapReduceChildJVM.java
index 528d2b78a2..c72b4009bd 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/MapReduceChildJVM.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/MapReduceChildJVM.java
@@ -19,6 +19,7 @@
 package org.apache.hadoop.mapred;
 
 import java.net.InetSocketAddress;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -77,7 +78,7 @@ public static void setVMEnv(Map<String, String> environment,
     // streaming) it will have the correct loglevel.
     environment.put(
         "HADOOP_ROOT_LOGGER", 
-        getChildLogLevel(conf, task.isMapTask()) + ",CLA"); 
+        getChildLogLevel(conf, task.isMapTask()) + ",console");
 
     // TODO: The following is useful for instance in streaming tasks. Should be
     // set in ApplicationMaster's env by the RM.
@@ -87,18 +88,19 @@ public static void setVMEnv(Map<String, String> environment,
     } else {
       hadoopClientOpts = hadoopClientOpts + " ";
     }
-    // FIXME: don't think this is also needed given we already set java
-    // properties.
-    long logSize = TaskLog.getTaskLogLength(conf);
-    Vector<String> logProps = new Vector<String>(4);
-    setupLog4jProperties(task, logProps, logSize);
-    Iterator<String> it = logProps.iterator();
-    StringBuffer buffer = new StringBuffer();
-    while (it.hasNext()) {
-      buffer.append(" " + it.next());
-    }
-    hadoopClientOpts = hadoopClientOpts + buffer.toString();
     environment.put("HADOOP_CLIENT_OPTS", hadoopClientOpts);
+    
+    // setEnvFromInputString above will add env variable values from
+    // mapredChildEnv to existing variables. We want to overwrite
+    // HADOOP_ROOT_LOGGER and HADOOP_CLIENT_OPTS if the user set it explicitly.
+    Map<String, String> tmpEnv = new HashMap<String, String>();
+    MRApps.setEnvFromInputString(tmpEnv, mapredChildEnv, conf);
+    String[] keys = { "HADOOP_ROOT_LOGGER", "HADOOP_CLIENT_OPTS" };
+    for (String key : keys) {
+      if (tmpEnv.containsKey(key)) {
+        environment.put(key, tmpEnv.get(key));
+      }
+    }
 
     // Add stdout/stderr env
     environment.put(
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestMapReduceChildJVM.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestMapReduceChildJVM.java
index 98094d5627..2b17236b91 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestMapReduceChildJVM.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestMapReduceChildJVM.java
@@ -18,11 +18,14 @@
 
 package org.apache.hadoop.mapreduce.v2.app.job.impl;
 
+import java.util.Map;
+
 import junit.framework.Assert;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapreduce.MRConfig;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.v2.api.records.JobState;
@@ -65,11 +68,20 @@ public void testCommandLine() throws Exception {
       " 0" +
       " 1><LOG_DIR>/stdout" +
       " 2><LOG_DIR>/stderr ]", app.myCommandLine);
+    
+    Assert.assertTrue("HADOOP_ROOT_LOGGER not set for job",
+      app.cmdEnvironment.containsKey("HADOOP_ROOT_LOGGER"));
+    Assert.assertEquals("INFO,console",
+      app.cmdEnvironment.get("HADOOP_ROOT_LOGGER"));
+    Assert.assertTrue("HADOOP_CLIENT_OPTS not set for job",
+      app.cmdEnvironment.containsKey("HADOOP_CLIENT_OPTS"));
+    Assert.assertEquals("", app.cmdEnvironment.get("HADOOP_CLIENT_OPTS"));
   }
 
   private static final class MyMRApp extends MRApp {
 
     private String myCommandLine;
+    private Map<String, String> cmdEnvironment;
 
     public MyMRApp(int maps, int reduces, boolean autoComplete,
         String testName, boolean cleanOnStart) {
@@ -88,10 +100,44 @@ public void handle(ContainerLauncherEvent event) {
             String cmdString = launchContext.getCommands().toString();
             LOG.info("launchContext " + cmdString);
             myCommandLine = cmdString;
+            cmdEnvironment = launchContext.getEnvironment();
           }
           super.handle(event);
         }
       };
     }
   }
+  
+  @Test
+  public void testEnvironmentVariables() throws Exception {
+    MyMRApp app = new MyMRApp(1, 0, true, this.getClass().getName(), true);
+    Configuration conf = new Configuration();
+    conf.set(JobConf.MAPRED_MAP_TASK_ENV, "HADOOP_CLIENT_OPTS=test");
+    conf.setStrings(MRJobConfig.MAP_LOG_LEVEL, "WARN");
+    conf.setBoolean(MRConfig.MAPREDUCE_APP_SUBMISSION_CROSS_PLATFORM, false);
+    Job job = app.submit(conf);
+    app.waitForState(job, JobState.SUCCEEDED);
+    app.verifyCompleted();
+    
+    Assert.assertTrue("HADOOP_ROOT_LOGGER not set for job",
+      app.cmdEnvironment.containsKey("HADOOP_ROOT_LOGGER"));
+    Assert.assertEquals("WARN,console",
+      app.cmdEnvironment.get("HADOOP_ROOT_LOGGER"));
+    Assert.assertTrue("HADOOP_CLIENT_OPTS not set for job",
+      app.cmdEnvironment.containsKey("HADOOP_CLIENT_OPTS"));
+    Assert.assertEquals("test", app.cmdEnvironment.get("HADOOP_CLIENT_OPTS"));
+
+    // Try one more.
+    app = new MyMRApp(1, 0, true, this.getClass().getName(), true);
+    conf = new Configuration();
+    conf.set(JobConf.MAPRED_MAP_TASK_ENV, "HADOOP_ROOT_LOGGER=trace");
+    job = app.submit(conf);
+    app.waitForState(job, JobState.SUCCEEDED);
+    app.verifyCompleted();
+    
+    Assert.assertTrue("HADOOP_ROOT_LOGGER not set for job",
+      app.cmdEnvironment.containsKey("HADOOP_ROOT_LOGGER"));
+    Assert.assertEquals("trace",
+      app.cmdEnvironment.get("HADOOP_ROOT_LOGGER"));
+  }
 }