diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index df2152e349..5f2c41cfea 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -436,6 +436,10 @@ Release 2.4.0 - UNRELEASED
     YARN-1793. Fixed ClientRMService#forceKillApplication not killing unmanaged
     application. (Karthik Kambatla via jianhe)
 
+    YARN-1788. Fixed a bug in ResourceManager to set the apps-completed and
+    apps-killed metrics correctly for killed applications. (Varun Vasudev via
+    vinodkv)
+
 Release 2.3.1 - UNRELEASED
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
index 8c9c52f239..0625708d18 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
@@ -1043,8 +1043,8 @@ public void transition(RMAppImpl app, RMAppEvent event) {
       if (app.finishTime == 0 ) {
         app.finishTime = System.currentTimeMillis();
       }
-      app.handler.handle(new AppRemovedSchedulerEvent(app.applicationId, app
-        .getState()));
+      app.handler.handle(new AppRemovedSchedulerEvent(app.applicationId,
+        finalState));
       app.handler.handle(
           new RMAppManagerEvent(app.applicationId,
           RMAppManagerEventType.APP_COMPLETED));
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRM.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRM.java
index b899ea708b..7747368dfe 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRM.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRM.java
@@ -62,6 +62,7 @@
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM;
 import org.apache.log4j.Level;
 import org.apache.log4j.LogManager;
@@ -519,7 +520,13 @@ protected Dispatcher createDispatcher() {
       }
     };
 
+    // test metrics
+    QueueMetrics metrics = rm.getResourceScheduler().getRootQueueMetrics();
+    int appsKilled = metrics.getAppsKilled();
+    int appsSubmitted = metrics.getAppsSubmitted();
+
     rm.start();
+    
     MockNM nm1 =
         new MockNM("127.0.0.1:1234", 15120, rm.getResourceTrackerService());
     nm1.registerNode();
@@ -552,6 +559,11 @@ protected Dispatcher createDispatcher() {
         new RMAppEvent(application.getApplicationId(),
           RMAppEventType.ATTEMPT_KILLED));
     rm.waitForState(application.getApplicationId(), RMAppState.KILLED);
+
+    // test metrics
+    metrics = rm.getResourceScheduler().getRootQueueMetrics();
+    Assert.assertEquals(appsKilled + 1, metrics.getAppsKilled());
+    Assert.assertEquals(appsSubmitted + 1, metrics.getAppsSubmitted());
   }
 
   public static void main(String[] args) throws Exception {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java
index 509bceb451..164d622ba5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestRMAppTransitions.java
@@ -60,6 +60,7 @@
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptUpdateSavedEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppRemovedSchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretManager;
@@ -87,6 +88,7 @@ public class TestRMAppTransitions {
   private RMStateStore store;
   private RMApplicationHistoryWriter writer;
   private YarnScheduler scheduler;
+  private TestSchedulerEventDispatcher schedulerDispatcher;
 
   // ignore all the RM application attempt events
   private static final class TestApplicationAttemptEventDispatcher implements
@@ -148,8 +150,11 @@ public void handle(RMAppManagerEvent event) {
   // handle all the scheduler events - same as in ResourceManager.java
   private static final class TestSchedulerEventDispatcher implements
       EventHandler<SchedulerEvent> {
+    public SchedulerEvent lastSchedulerEvent;
+    
     @Override
     public void handle(SchedulerEvent event) {
+      lastSchedulerEvent = event;
     }
   }  
 
@@ -201,8 +206,9 @@ null, new AMRMTokenSecretManager(conf),
     rmDispatcher.register(RMAppManagerEventType.class,
         new TestApplicationManagerEventDispatcher());
     
+    schedulerDispatcher = new TestSchedulerEventDispatcher();
     rmDispatcher.register(SchedulerEventType.class,
-        new TestSchedulerEventDispatcher());
+        schedulerDispatcher);
     
     rmDispatcher.init(conf);
     rmDispatcher.start();
@@ -502,6 +508,7 @@ public void testAppNewKill() throws IOException {
     assertKilled(application);
     assertAppFinalStateNotSaved(application);
     verifyApplicationFinished(RMAppState.KILLED);
+    verifyAppRemovedSchedulerEvent(RMAppState.KILLED);
   }
 
   @Test
@@ -534,6 +541,7 @@ public void testAppNewSavingKill() throws IOException {
     sendAppUpdateSavedEvent(application);
     assertKilled(application);
     verifyApplicationFinished(RMAppState.KILLED);
+    verifyAppRemovedSchedulerEvent(RMAppState.KILLED);
   }
 
   @Test (timeout = 30000)
@@ -583,6 +591,7 @@ public void testAppSubmittedKill() throws IOException, InterruptedException {
     assertKilled(application);
     assertAppFinalStateSaved(application);
     verifyApplicationFinished(RMAppState.KILLED);
+    verifyAppRemovedSchedulerEvent(RMAppState.KILLED);
   }
 
   @Test
@@ -640,6 +649,7 @@ public void testAppAcceptedKill() throws IOException, InterruptedException {
     assertKilled(application);
     assertAppFinalStateSaved(application);
     verifyApplicationFinished(RMAppState.KILLED);
+    verifyAppRemovedSchedulerEvent(RMAppState.KILLED);
   }
 
   @Test
@@ -663,6 +673,7 @@ public void testAppRunningKill() throws IOException {
     sendAppUpdateSavedEvent(application);
     assertKilled(application);
     verifyApplicationFinished(RMAppState.KILLED);
+    verifyAppRemovedSchedulerEvent(RMAppState.KILLED);
   }
 
   @Test
@@ -868,4 +879,15 @@ private void verifyApplicationFinished(RMAppState state) {
     verify(writer).applicationFinished(any(RMApp.class), finalState.capture());
     Assert.assertEquals(state, finalState.getValue());
   }
+  
+  private void verifyAppRemovedSchedulerEvent(RMAppState finalState) {
+    Assert.assertEquals(SchedulerEventType.APP_REMOVED,
+      schedulerDispatcher.lastSchedulerEvent.getType());
+    if(schedulerDispatcher.lastSchedulerEvent instanceof 
+        AppRemovedSchedulerEvent) {
+      AppRemovedSchedulerEvent appRemovedEvent =
+          (AppRemovedSchedulerEvent) schedulerDispatcher.lastSchedulerEvent;
+      Assert.assertEquals(finalState, appRemovedEvent.getFinalState());
+    }
+  }
 }