YARN-7663. RMAppImpl:Invalid event: START at KILLED. Contributed by lujie

(cherry picked from commit d2c2d8c913dfc12585394a1168d4a92934a4cfb4)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java

@@ -393,7 +393,7 @@ public class RMAppImpl implements RMApp, Recoverable {
         EnumSet.of(RMAppEventType.APP_ACCEPTED,
             RMAppEventType.APP_REJECTED, RMAppEventType.KILL,
             RMAppEventType.ATTEMPT_FINISHED, RMAppEventType.ATTEMPT_FAILED,
-            RMAppEventType.NODE_UPDATE))
+            RMAppEventType.NODE_UPDATE, RMAppEventType.START))
 
      .installTopology();
 
@@ -879,7 +879,7 @@ public class RMAppImpl implements RMApp, Recoverable {
       } catch (InvalidStateTransitionException e) {
         LOG.error("App: " + appID
             + " can't handle this event at current state", e);
-        /* TODO fail the application on the failed transition */
+        onInvalidStateTransition(event.getType(), oldState);
       }
 
       // Log at INFO if we're not recovering or not in a terminal state.
@@ -2014,4 +2014,14 @@ public class RMAppImpl implements RMApp, Recoverable {
     this.submissionContext.setAMContainerSpec(null);
     this.submissionContext.setLogAggregationContext(null);
   }
+
+  /**
+     * catch the InvalidStateTransition.
+     * @param state
+     * @param rmAppEventType
+     */
+  protected void onInvalidStateTransition(RMAppEventType rmAppEventType,
+              RMAppState state){
+      /* TODO fail the application on the failed transition */
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java

@@ -1151,6 +1151,41 @@ public class TestRMAppTransitions {
     verifyRMAppFieldsForFinalTransitions(application);
   }
   
+  @Test (timeout = 30000)
+  public void testAppStartAfterKilled() throws IOException {
+    LOG.info("--- START: testAppStartAfterKilled ---");
+
+    ApplicationId applicationId = MockApps.newAppID(appId++);
+    RMApp application = new RMAppImpl(applicationId, rmContext, conf, null,
+            null, null, new ApplicationSubmissionContextPBImpl(), null, null,
+              System.currentTimeMillis(), "YARN", null, null) {
+          @Override
+          protected void onInvalidStateTransition(RMAppEventType rmAppEventType,
+                  RMAppState state) {
+            Assert.assertTrue("RMAppImpl: can't handle " + rmAppEventType
+                                 + " at state " + state, false);
+          }
+      };
+
+    // NEW => KILLED event RMAppEventType.KILL
+    UserGroupInformation fooUser = UserGroupInformation.createUserForTesting(
+            "testAppStartAfterKilled", new String[] {"foo_group"});
+    RMAppEvent event = new RMAppKillByClientEvent(
+             applicationId, "Application killed by user.",
+             fooUser, Server.getRemoteIp());
+    application.handle(event);
+    rmDispatcher.await();
+    assertKilled(application);
+
+    // KILLED => KILLED event RMAppEventType.START
+    event = new RMAppFailedAttemptEvent(application.getApplicationId(),
+                            RMAppEventType.START, "", false);
+    application.handle(event);
+    rmDispatcher.await();
+    assertTimesAtFinish(application);
+    assertAppState(RMAppState.KILLED, application);
+  }
+
   @Test(timeout = 30000)
   public void testAppsRecoveringStates() throws Exception {
     RMState state = new RMState();