diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 5eaf4f4efac..03bb20bebf4 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -692,6 +692,9 @@ Release 2.7.0 - UNRELEASED
     YARN-3265. Fixed a deadlock in CapacityScheduler by always passing a queue's
     available resource-limit from the parent queue. (Wangda Tan via vinodkv)
 
+    YARN-3131. YarnClientImpl should check FAILED and KILLED state in
+    submitApplication (Chang Li via jlowe)
+
 Release 2.6.0 - 2014-11-18
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java
index 6acf7d80857..d6b36bbf8dd 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java
@@ -254,13 +254,22 @@ public class YarnClientImpl extends YarnClient {
 
     int pollCount = 0;
     long startTime = System.currentTimeMillis();
-
+    EnumSet<YarnApplicationState> waitingStates = 
+                                 EnumSet.of(YarnApplicationState.NEW,
+                                 YarnApplicationState.NEW_SAVING,
+                                 YarnApplicationState.SUBMITTED);
+    EnumSet<YarnApplicationState> failToSubmitStates = 
+                                  EnumSet.of(YarnApplicationState.FAILED,
+                                  YarnApplicationState.KILLED);		
     while (true) {
       try {
-        YarnApplicationState state =
-            getApplicationReport(applicationId).getYarnApplicationState();
-        if (!state.equals(YarnApplicationState.NEW) &&
-            !state.equals(YarnApplicationState.NEW_SAVING)) {
+        ApplicationReport appReport = getApplicationReport(applicationId);
+        YarnApplicationState state = appReport.getYarnApplicationState();
+        if (!waitingStates.contains(state)) {
+          if(failToSubmitStates.contains(state)) {
+            throw new YarnException("Failed to submit " + applicationId + 
+                " to YARN : " + appReport.getDiagnostics());
+          }
           LOG.info("Submitted application " + applicationId);
           break;
         }
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/ProtocolHATestBase.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/ProtocolHATestBase.java
index da7d50529ac..782bc43e51a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/ProtocolHATestBase.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/ProtocolHATestBase.java
@@ -646,7 +646,7 @@ public abstract class ProtocolHATestBase extends ClientBaseWithFixes {
       ApplicationReport report =
           ApplicationReport.newInstance(appId, attemptId, "fakeUser",
               "fakeQueue", "fakeApplicationName", "localhost", 0, null,
-              YarnApplicationState.FAILED, "fake an application report", "",
+              YarnApplicationState.FINISHED, "fake an application report", "",
               1000l, 1200l, FinalApplicationStatus.FAILED, null, "", 50f,
               "fakeApplicationType", null);
       return report;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java
index 7e97134437c..9946506f413 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestYarnClient.java
@@ -157,12 +157,9 @@ public class TestYarnClient {
 
     YarnApplicationState[] exitStates = new YarnApplicationState[]
         {
-          YarnApplicationState.SUBMITTED,
           YarnApplicationState.ACCEPTED,
           YarnApplicationState.RUNNING,
-          YarnApplicationState.FINISHED,
-          YarnApplicationState.FAILED,
-          YarnApplicationState.KILLED
+          YarnApplicationState.FINISHED
         };
 
     // Submit an application without ApplicationId provided
@@ -203,6 +200,54 @@ public class TestYarnClient {
     client.stop();
   }
 
+  @Test (timeout = 30000)
+  public void testSubmitIncorrectQueue() throws IOException {
+    MiniYARNCluster cluster = new MiniYARNCluster("testMRAMTokens", 1, 1, 1);
+    YarnClient rmClient = null;
+    try {
+      cluster.init(new YarnConfiguration());
+	     cluster.start();
+      final Configuration yarnConf = cluster.getConfig();
+      rmClient = YarnClient.createYarnClient();
+      rmClient.init(yarnConf);
+      rmClient.start();
+      YarnClientApplication newApp = rmClient.createApplication();
+
+      ApplicationId appId = newApp.getNewApplicationResponse().getApplicationId();
+
+      // Create launch context for app master
+      ApplicationSubmissionContext appContext
+        = Records.newRecord(ApplicationSubmissionContext.class);
+
+      // set the application id
+      appContext.setApplicationId(appId);
+
+      // set the application name
+      appContext.setApplicationName("test");
+
+      // Set the queue to which this application is to be submitted in the RM
+      appContext.setQueue("nonexist");
+
+      // Set up the container launch context for the application master
+      ContainerLaunchContext amContainer
+        = Records.newRecord(ContainerLaunchContext.class);
+      appContext.setAMContainerSpec(amContainer);
+      appContext.setResource(Resource.newInstance(1024, 1));
+      // appContext.setUnmanagedAM(unmanaged);
+
+      // Submit the application to the applications manager
+      rmClient.submitApplication(appContext);
+      Assert.fail("Job submission should have thrown an exception");
+    } catch (YarnException e) {
+      Assert.assertTrue(e.getMessage().contains("Failed to submit"));
+    } finally {
+      if (rmClient != null) {
+        rmClient.stop();
+      }
+      cluster.stop();
+    }
+  }
+  
   @Test
   public void testKillApplication() throws Exception {
     MockRM rm = new MockRM();
@@ -998,7 +1043,7 @@ public class TestYarnClient {
       public ApplicationReport getApplicationReport(ApplicationId appId) {
         ApplicationReport report = mock(ApplicationReport.class);
         when(report.getYarnApplicationState())
-            .thenReturn(YarnApplicationState.SUBMITTED);
+            .thenReturn(YarnApplicationState.RUNNING);
         return report;
       }