TestRM/MockRM works!!

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/MR-279@1153431 13f79535-47bb-0310-9956-ffa450edef68

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ams/ApplicationMasterService.java

@@ -99,6 +99,8 @@ AMRMProtocol, EventHandler<ApplicationMasterServiceEvent> {
       conf.get(YarnConfiguration.SCHEDULER_ADDRESS,
           YarnConfiguration.DEFAULT_SCHEDULER_BIND_ADDRESS);
     masterServiceAddress =  NetUtils.createSocketAddr(bindAddress);
+    this.rmContext.getDispatcher().register(
+        ApplicationMasterServiceEventType.class, this);
     super.init(conf);
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java

@@ -24,6 +24,8 @@ import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEventType;
+import org.apache.hadoop.yarn.server.resourcemanager.ams.ApplicationMasterServiceEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.ams.ApplicationMasterServiceEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent;
@@ -102,7 +104,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
 
        // Transitions from NEW State
       .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.SUBMITTED,
-          RMAppAttemptEventType.START, new AddToSchedulerTransition())
+          RMAppAttemptEventType.START, new AttemptStartedTransition())
       .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.KILLED,
           RMAppAttemptEventType.KILL)
 
@@ -113,7 +115,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
           RMAppAttemptEventType.APP_ACCEPTED, new ScheduleTransition())
       .addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.KILLED,
           RMAppAttemptEventType.KILL,
-          new BaseFinalTransition(RMAppEventType.ATTEMPT_KILLED))
+          new BaseFinalTransition(RMAppAttemptState.KILLED))
 
        // Transitions from SCHEDULED State
       .addTransition(RMAppAttemptState.SCHEDULED,
@@ -122,7 +124,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
           new AMContainerAllocatedTransition())
       .addTransition(RMAppAttemptState.SCHEDULED, RMAppAttemptState.KILLED,
           RMAppAttemptEventType.KILL,
-          new BaseFinalTransition(RMAppEventType.ATTEMPT_KILLED))
+          new BaseFinalTransition(RMAppAttemptState.KILLED))
 
        // Transitions from ALLOCATED State
       .addTransition(RMAppAttemptState.ALLOCATED, RMAppAttemptState.LAUNCHED,
@@ -141,10 +143,10 @@ public class RMAppAttemptImpl implements RMAppAttempt {
       .addTransition(
           RMAppAttemptState.LAUNCHED, RMAppAttemptState.FAILED,
           RMAppAttemptEventType.EXPIRE,
-          new FinalTransition(RMAppEventType.ATTEMPT_FAILED))
+          new FinalTransition(RMAppAttemptState.FAILED))
       .addTransition(RMAppAttemptState.LAUNCHED, RMAppAttemptState.KILLED,
           RMAppAttemptEventType.KILL,
-          new FinalTransition(RMAppEventType.ATTEMPT_KILLED))
+          new FinalTransition(RMAppAttemptState.KILLED))
 
        // Transitions from RUNNING State
       .addTransition(RMAppAttemptState.RUNNING, RMAppAttemptState.FINISHED,
@@ -162,11 +164,11 @@ public class RMAppAttemptImpl implements RMAppAttempt {
       .addTransition(
           RMAppAttemptState.RUNNING, RMAppAttemptState.FAILED,
           RMAppAttemptEventType.EXPIRE,
-          new FinalTransition(RMAppEventType.ATTEMPT_FAILED))
+          new FinalTransition(RMAppAttemptState.FAILED))
       .addTransition(
           RMAppAttemptState.RUNNING, RMAppAttemptState.KILLED,
           RMAppAttemptEventType.KILL,
-          new FinalTransition(RMAppEventType.ATTEMPT_KILLED))
+          new FinalTransition(RMAppAttemptState.KILLED))
 
       // Transitions from FAILED State
       .addTransition(RMAppAttemptState.FAILED, RMAppAttemptState.FAILED,
@@ -389,11 +391,16 @@ public class RMAppAttemptImpl implements RMAppAttempt {
     
   }
 
-  private static final class AddToSchedulerTransition extends BaseTransition {
+  private static final class AttemptStartedTransition extends BaseTransition {
     @Override
     public void transition(RMAppAttemptImpl appAttempt,
         RMAppAttemptEvent event) {
 
+      // Register with the ApplicationMasterService
+      appAttempt.eventHandler.handle(new ApplicationMasterServiceEvent(
+          appAttempt.applicationAttemptId,
+          ApplicationMasterServiceEventType.REGISTER));
+
       // Add the application to the scheduler
       appAttempt.eventHandler.handle(
           new AppAddedSchedulerEvent(appAttempt.applicationAttemptId,
@@ -455,22 +462,42 @@ public class RMAppAttemptImpl implements RMAppAttempt {
 
   private static class BaseFinalTransition extends BaseTransition {
 
-    private final RMAppEventType outgoingEvent;
+    private final RMAppAttemptState finalAttemptState;
 
-    public BaseFinalTransition(RMAppEventType outgoingEventToApp) {
-      this.outgoingEvent = outgoingEventToApp;
+    public BaseFinalTransition(RMAppAttemptState finalAttemptState) {
+      this.finalAttemptState = finalAttemptState;
     }
 
     @Override
     public void transition(RMAppAttemptImpl appAttempt,
         RMAppAttemptEvent event) {
 
+      // Tell the AMS
+      // Register with the ApplicationMasterService
+      appAttempt.eventHandler.handle(new ApplicationMasterServiceEvent(
+          appAttempt.applicationAttemptId,
+          ApplicationMasterServiceEventType.UNREGISTER));
+
       // Tell the application and the scheduler
+      RMAppEventType eventToApp = null;
+      switch (finalAttemptState) {
+      case FINISHED:
+        eventToApp = RMAppEventType.ATTEMPT_FINISHED;
+        break;
+      case KILLED:
+        eventToApp = RMAppEventType.ATTEMPT_KILLED;
+        break;
+      case FAILED:
+        eventToApp = RMAppEventType.ATTEMPT_FAILED;
+        break;
+      default:
+        LOG.info("Cannot get this state!! Error!!");
+        break;
+      }
       appAttempt.eventHandler.handle(new RMAppEvent(
-          appAttempt.applicationAttemptId.getApplicationId(),
-          this.outgoingEvent));
+          appAttempt.applicationAttemptId.getApplicationId(), eventToApp));
       appAttempt.eventHandler.handle(new AppRemovedSchedulerEvent(appAttempt
-          .getAppAttemptId()));
+          .getAppAttemptId(), finalAttemptState));
     }
   }
 
@@ -489,7 +516,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
   private static final class LaunchFailedTransition extends BaseFinalTransition {
 
     public LaunchFailedTransition() {
-      super(RMAppEventType.ATTEMPT_FAILED);
+      super(RMAppAttemptState.FAILED);
     }
 
     @Override
@@ -510,7 +537,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
   private static final class KillAllocatedAMTransition extends
       BaseFinalTransition {
     public KillAllocatedAMTransition() {
-      super(RMAppEventType.ATTEMPT_KILLED);
+      super(RMAppAttemptState.KILLED);
     }
 
     @Override
@@ -549,7 +576,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
       BaseFinalTransition {
 
     public AMContainerCrashedTransition() {
-      super(RMAppEventType.ATTEMPT_FAILED);
+      super(RMAppAttemptState.FAILED);
     }
 
     @Override
@@ -571,8 +598,8 @@ public class RMAppAttemptImpl implements RMAppAttempt {
 
   private static class FinalTransition extends BaseFinalTransition {
 
-    public FinalTransition(RMAppEventType outgoingEventToApp) {
-      super(outgoingEventToApp);
+    public FinalTransition(RMAppAttemptState finalAttemptState) {
+      super(finalAttemptState);
     }
 
     @Override
@@ -636,7 +663,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
   private static final class AMUnregisteredTransition extends FinalTransition {
 
     public AMUnregisteredTransition() {
-      super(RMAppEventType.ATTEMPT_FINISHED);
+      super(RMAppAttemptState.FINISHED);
     }
 
     @Override
@@ -671,7 +698,7 @@ public class RMAppAttemptImpl implements RMAppAttempt {
       // the AMContainer, AppAttempt fails
       if (appAttempt.masterContainer.getId().equals(
           containerId)) {
-        new BaseFinalTransition(RMAppEventType.ATTEMPT_FAILED).transition(
+        new BaseFinalTransition(RMAppAttemptState.FAILED).transition(
             appAttempt, containerFinishedEvent);
         return RMAppAttemptState.FAILED;
       }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java

@@ -45,6 +45,7 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.ApplicationsStore.ApplicationStore;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 
 /**
@@ -78,17 +79,14 @@ public class AppSchedulingInfo {
   List<Container> completedContainers = new ArrayList<Container>();
   /* Allocated by scheduler */
   List<Container> allocated = new ArrayList<Container>();
-  ApplicationMaster master;
   boolean pending = true; // for app metrics
 
   public AppSchedulingInfo(ApplicationAttemptId appAttemptId,
-      ApplicationMaster master, String queueName, String user,
-      ApplicationStore store) {
+      String queueName, String user, ApplicationStore store) {
     this.applicationAttemptId = appAttemptId;
     this.applicationId = appAttemptId.getApplicationId();
     this.queueName = queueName;
     this.user = user;
-    this.master = master;
     this.store = store;
   }
 
@@ -108,10 +106,6 @@ public class AppSchedulingInfo {
     return user;
   }
 
-  public synchronized ApplicationState getState() {
-    return master.getState();
-  }
-
   public synchronized boolean isPending() {
     return pending;
   }
@@ -390,7 +384,7 @@ public class AppSchedulingInfo {
     }
   }
 
-  synchronized public void stop() {
+  synchronized public void stop(RMAppAttemptState rmAppAttemptFinalState) {
     // clear pending resources metrics for the application
     QueueMetrics metrics = queue.getMetrics();
     for (Map<String, ResourceRequest> asks : requests.values()) {
@@ -401,7 +395,7 @@ public class AppSchedulingInfo {
                 .getNumContainers()));
       }
     }
-    metrics.finishApp(this);
+    metrics.finishApp(this, rmAppAttemptFinalState);
     
     // Clear requests themselves
     clearRequests();

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java

@@ -16,6 +16,7 @@ import org.apache.hadoop.metrics2.lib.MutableCounterInt;
 import org.apache.hadoop.metrics2.lib.MutableGaugeInt;
 import org.apache.hadoop.yarn.api.records.ApplicationState;
 import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
 import org.apache.hadoop.yarn.util.Self;
 import static org.apache.hadoop.yarn.server.resourcemanager.resource.Resources.*;
 
@@ -133,9 +134,9 @@ public class QueueMetrics {
     }
   }
 
-  public void finishApp(AppSchedulingInfo app) {
-    ApplicationState state = app.getState();
-    switch (state) {
+  public void finishApp(AppSchedulingInfo app,
+      RMAppAttemptState rmAppAttemptFinalState) {
+    switch (rmAppAttemptFinalState) {
       case KILLED: appsKilled.incr(); break;
       case FAILED: appsFailed.incr(); break;
       default: appsCompleted.incr();  break;
@@ -147,10 +148,10 @@ public class QueueMetrics {
     }
     QueueMetrics userMetrics = getUserMetrics(app.getUser());
     if (userMetrics != null) {
-      userMetrics.finishApp(app);
+      userMetrics.finishApp(app, rmAppAttemptFinalState);
     }
     if (parent != null) {
-      parent.finishApp(app);
+      parent.finishApp(app, rmAppAttemptFinalState);
     }
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApp.java

@@ -17,6 +17,7 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 
 public class SchedulerApp {
@@ -98,8 +99,8 @@ public class SchedulerApp {
     return this.queue;
   }
 
-  public void stop() {
-    this.appSchedulingInfo.stop();
+  public void stop(RMAppAttemptState rmAppAttemptFinalState) {
+    this.appSchedulingInfo.stop(rmAppAttemptFinalState);
   }
 
   synchronized public void completedContainer(Container container, 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java

@@ -58,6 +58,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
 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.rmapp.attempt.event.RMAppAttemptRejectedEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AppSchedulingInfo;
@@ -308,7 +309,7 @@ implements ResourceScheduler, CapacitySchedulerContext {
     }
 
     AppSchedulingInfo appSchedulingInfo = new AppSchedulingInfo(
-        applicationAttemptId, null, queueName, user, null);
+        applicationAttemptId, queueName, user, null);
     CSApp csApp = new CSApp(appSchedulingInfo, queue);
 
     // Submit to the queue
@@ -334,9 +335,10 @@ implements ResourceScheduler, CapacitySchedulerContext {
   }
 
   private synchronized void doneApplication(
-      ApplicationAttemptId applicationAttemptId, boolean finishApplication) {
+      ApplicationAttemptId applicationAttemptId,
+      RMAppAttemptState rmAppAttemptFinalState) {
     LOG.info("Application " + applicationAttemptId + " is done." +
-    		" finish=" + finishApplication);
+    		" finalState=" + rmAppAttemptFinalState);
     
     CSApp application = getApplication(applicationAttemptId);
 
@@ -354,25 +356,20 @@ implements ResourceScheduler, CapacitySchedulerContext {
     releaseReservedContainers(application);
     
     // Clean up pending requests, metrics etc.
-    application.stop();
+    application.stop(rmAppAttemptFinalState);
     
-    /** The application can be retried. So only remove it from scheduler data
-     * structures if the finishApplication flag is set.
-     */
-    if (finishApplication) {
-      // Inform the queue
-      String queueName = application.getQueue().getQueueName();
-      Queue queue = queues.get(queueName);
-      if (!(queue instanceof LeafQueue)) {
-        LOG.error("Cannot finish application " + "from non-leaf queue: "
-            + queueName);
-      } else {
-        queue.finishApplication(application, queue.getQueueName());
-      }
-      
-      // Remove from our data-structure
-      applications.remove(applicationAttemptId);
+    // Inform the queue
+    String queueName = application.getQueue().getQueueName();
+    Queue queue = queues.get(queueName);
+    if (!(queue instanceof LeafQueue)) {
+      LOG.error("Cannot finish application " + "from non-leaf queue: "
+          + queueName);
+    } else {
+      queue.finishApplication(application, queue.getQueueName());
     }
+    
+    // Remove from our data-structure
+    applications.remove(applicationAttemptId);
   }
 
   @Override
@@ -621,7 +618,8 @@ implements ResourceScheduler, CapacitySchedulerContext {
       break;
     case APP_REMOVED:
       AppRemovedSchedulerEvent appRemovedEvent = (AppRemovedSchedulerEvent)event;
-      doneApplication(appRemovedEvent.getApplicationAttemptID(), true);
+      doneApplication(appRemovedEvent.getApplicationAttemptID(),
+          appRemovedEvent.getFinalAttemptState());
       break;
     case CONTAINER_FINISHED:
       ContainerFinishedSchedulerEvent containerFinishedEvent = (ContainerFinishedSchedulerEvent) event;

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/event/AppRemovedSchedulerEvent.java

@@ -1,19 +1,25 @@
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.event;
 
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
 
 public class AppRemovedSchedulerEvent extends SchedulerEvent {
 
   private final ApplicationAttemptId applicationAttemptId;
+  private final RMAppAttemptState finalAttemptState;
 
-  public AppRemovedSchedulerEvent(ApplicationAttemptId applicationAttemptId) {
+  public AppRemovedSchedulerEvent(ApplicationAttemptId applicationAttemptId,
+      RMAppAttemptState finalAttemptState) {
     super(SchedulerEventType.APP_REMOVED);
     this.applicationAttemptId = applicationAttemptId;
+    this.finalAttemptState = finalAttemptState;
   }
 
   public ApplicationAttemptId getApplicationAttemptID() {
     return this.applicationAttemptId;
   }
 
+  public RMAppAttemptState getFinalAttemptState() {
+    return this.finalAttemptState;
+  }
 }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java

@@ -60,6 +60,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.recovery.Store.RMState;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
 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.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
@@ -262,7 +263,7 @@ public class FifoScheduler implements ResourceScheduler {
   private synchronized void addApplication(ApplicationAttemptId appAttemptId,
       String queueName, String user) {
     AppSchedulingInfo appSchedulingInfo = new AppSchedulingInfo(
-        appAttemptId, null, queueName, user, null);
+        appAttemptId, queueName, user, null);
     SchedulerApp schedulerApp = new SchedulerApp(appSchedulingInfo,
         DEFAULT_QUEUE);
     applications.put(appAttemptId, schedulerApp);
@@ -275,7 +276,8 @@ public class FifoScheduler implements ResourceScheduler {
   }
 
   private synchronized void doneApplication(
-      ApplicationAttemptId applicationAttemptId, boolean finishApplication)
+      ApplicationAttemptId applicationAttemptId,
+      RMAppAttemptState rmAppAttemptFinalState)
       throws IOException {
     SchedulerApp application = getApplication(applicationAttemptId);
     if (application == null) {
@@ -284,12 +286,10 @@ public class FifoScheduler implements ResourceScheduler {
     }
 
     // Clean up pending requests, metrics etc.
-    application.stop();
-    
-    if (finishApplication) {
-      // Remove the application
-      applications.remove(applicationAttemptId);
-    }
+    application.stop(rmAppAttemptFinalState);
+
+    // Remove the application
+    applications.remove(applicationAttemptId);
   }
 
   /**
@@ -548,8 +548,6 @@ public class FifoScheduler implements ResourceScheduler {
     applicationCompletedContainers(getCompletedContainers(containers));
 
     LOG.info("Node heartbeat " + rmNode.getNodeID() + " resource = " + node.getAvailableResource());
-    LOG.info("Node heartbeat " + rmNode.getNodeID() + " resource = " + node.getAvailableResource().getMemory());
-    LOG.info("=========Node heartbeat " + rmNode.getNodeID() + " resourcemimi = " + minimumAllocation.getMemory());
     if (Resources.greaterThanOrEqual(node.getAvailableResource(),
         minimumAllocation)) {
       assignContainers(node);
@@ -590,7 +588,8 @@ public class FifoScheduler implements ResourceScheduler {
     case APP_REMOVED:
       AppRemovedSchedulerEvent appRemovedEvent = (AppRemovedSchedulerEvent)event;
       try {
-        doneApplication(appRemovedEvent.getApplicationAttemptID(), true);
+        doneApplication(appRemovedEvent.getApplicationAttemptID(),
+            appRemovedEvent.getFinalAttemptState());
       } catch(IOException ie) {
         LOG.error("Unable to remove application "
             + appRemovedEvent.getApplicationAttemptID(), ie);

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRM.java

@@ -244,6 +244,10 @@ public class MockRM extends ResourceManager {
       public void start() {
         //override to not start rpc handler
       }
+      @Override
+      public void stop() {
+        // don't do anything
+      }
     };
   }
 
@@ -255,6 +259,10 @@ public class MockRM extends ResourceManager {
       public void start() {
         //override to not start rpc handler
       }
+      @Override
+      public void stop() {
+        // don't do anything
+      }
     };
   }
 
@@ -266,6 +274,10 @@ public class MockRM extends ResourceManager {
       public void start() {
         //override to not start rpc handler
       }
+      @Override
+      public void stop() {
+        // don't do anything
+      }
     };
   }
 
@@ -282,6 +294,10 @@ public class MockRM extends ResourceManager {
       public void  handle(AMLauncherEvent appEvent) {
         //don't do anything
       }
+      @Override
+      public void stop() {
+        // don't do anything
+      }
     };
   }
 
@@ -292,6 +308,10 @@ public class MockRM extends ResourceManager {
       public void start() {
         //override to not start rpc handler
       }
+      @Override
+      public void stop() {
+        // don't do anything
+      }
     };
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRM.java

@@ -29,6 +29,8 @@ public class TestRM {
     rm.registerAppAttempt(attempt.getAppAttemptId());
     rm.unregisterAppAttempt(attempt.getAppAttemptId());
     rm.waitForState(attempt.getAppAttemptId(), RMAppAttemptState.FINISHED);
+
+    rm.stop();
   }
 
   public static void main(String[] args) throws Exception {