diff --git a/hadoop-mapreduce-project/CHANGES.txt b/hadoop-mapreduce-project/CHANGES.txt
index bcc1c5ba84..69296805d0 100644
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@@ -1552,6 +1552,9 @@ Release 0.23.0 - Unreleased
     MAPREDUCE-3153. Fix TestFileOutputCommitter which was broken by
     MAPREDUCE-2702. (mahadev via acmurthy) 
 
+    MAPREDUCE-3123. Fix NM to quote symlink names to escape special
+    characters. (Hitesh Shah via acmurthy) 
+
 Release 0.22.0 - Unreleased
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java b/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
index 117ede79a1..a4c9d625c4 100644
--- a/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
+++ b/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
@@ -268,7 +268,7 @@ public ShellScriptBuilder symlink(Path src, Path dst) throws IOException {
       if (dst.toUri().getPath().indexOf('/') != -1) {
         line("mkdir -p ", dst.getParent().toString());
       }
-      line("ln -sf ", src.toUri().getPath(), " ", dst.toString());
+      line("ln -sf \"", src.toUri().getPath(), "\" \"", dst.toString(), "\"");
       return this;
     }
   
@@ -349,8 +349,8 @@ public void sanitizeEnv(Map<String, String> environment,
     putEnvIfAbsent(environment, Environment.YARN_HOME.name());
 
   }
-  
-  private static void writeLaunchEnv(OutputStream out,
+    
+  static void writeLaunchEnv(OutputStream out,
       Map<String,String> environment, Map<Path,String> resources,
       List<String> command)
       throws IOException {
diff --git a/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerLaunch.java b/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerLaunch.java
new file mode 100644
index 0000000000..8fc35158bc
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerLaunch.java
@@ -0,0 +1,100 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher;
+
+import static org.junit.Assert.*;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
+import org.junit.Test;
+
+public class TestContainerLaunch {
+
+  @Test
+  public void testSpecialCharSymlinks() throws IOException  {
+
+    String rootDir = new File(System.getProperty(
+        "test.build.data", "/tmp")).getAbsolutePath();
+    File shellFile = null;
+    File tempFile = null;
+    String badSymlink = "foo@zz%_#*&!-+= bar()";
+    File symLinkFile = null;
+
+    try {
+      shellFile = new File(rootDir, "hello.sh");
+      tempFile = new File(rootDir, "temp.sh");
+      String timeoutCommand = "echo \"hello\"";
+      PrintWriter writer = new PrintWriter(new FileOutputStream(shellFile));    
+      shellFile.setExecutable(true);
+      writer.println(timeoutCommand);
+      writer.close();
+
+      Map<Path, String> resources = new HashMap<Path, String>();
+      Path path = new Path(shellFile.getAbsolutePath());
+      resources.put(path, badSymlink);
+
+      FileOutputStream fos = new FileOutputStream(tempFile);
+
+      Map<String, String> env = new HashMap<String, String>();
+      List<String> commands = new ArrayList<String>();
+      commands.add("/bin/sh ./\\\"" + badSymlink + "\\\"");
+
+      ContainerLaunch.writeLaunchEnv(fos, env, resources, commands);
+      fos.flush();
+      fos.close();
+      tempFile.setExecutable(true);
+
+      File rootDirFile = new File(rootDir);
+      Shell.ShellCommandExecutor shexc 
+      = new Shell.ShellCommandExecutor(new String[]{tempFile.getAbsolutePath()}, rootDirFile);
+
+      shexc.execute();
+      assertEquals(shexc.getExitCode(), 0);
+      assert(shexc.getOutput().contains("hello"));
+
+      symLinkFile = new File(rootDir, badSymlink);      
+    }
+    finally {
+      // cleanup
+      if (shellFile != null
+          && shellFile.exists()) {
+        shellFile.delete();
+      }
+      if (tempFile != null 
+          && tempFile.exists()) {
+        tempFile.delete();
+      }
+      if (symLinkFile != null
+          && symLinkFile.exists()) {
+        symLinkFile.delete();
+      } 
+    }
+  }
+
+}