@Test(timeout=4000)
public void testConcurrentAppSubmit()
    throws IOException, InterruptedException, BrokenBarrierException,
    YarnException {

  ResourceScheduler scheduler = mockResourceScheduler();
  RMContext rmContext = mock(RMContext.class);
  mockRMContext(scheduler, rmContext);
  RMStateStore stateStore = mock(RMStateStore.class);
  when(rmContext.getStateStore()).thenReturn(stateStore);
  RMAppManager appManager = new RMAppManager(rmContext, scheduler,
      null, mock(ApplicationACLsManager.class), new Configuration());
  final ApplicationId appId1 = getApplicationId(100);
  final ApplicationId appId2 = getApplicationId(101);
  final SubmitApplicationRequest submitRequest1 = mockSubmitAppRequest(
      appId1, null, null);
  final SubmitApplicationRequest submitRequest2 = mockSubmitAppRequest(
      appId2, null, null);
  
  final CyclicBarrier startBarrier = new CyclicBarrier(2);
  final CyclicBarrier endBarrier = new CyclicBarrier(2);
  EventHandler<Event> eventHandler = new EventHandler<Event>() {
    @Override
    public void handle(Event rawEvent) {
      if (rawEvent instanceof RMAppEvent) {
        RMAppEvent event = (RMAppEvent) rawEvent;
        if (event.getApplicationId().equals(appId1)) {
          try {
            startBarrier.await();
            endBarrier.await();
          } catch (BrokenBarrierException e) {
            LOG.warn("Broken Barrier", e);
          } catch (InterruptedException e) {
            LOG.warn("Interrupted while awaiting barriers", e);
          }
        }
      }
    }
  };
  when(rmContext.getDispatcher().getEventHandler()).thenReturn(eventHandler);
  doReturn(mock(RMTimelineCollectorManager.class)).when(rmContext)
      .getRMTimelineCollectorManager();
  final ClientRMService rmService =
      new ClientRMService(rmContext, scheduler, appManager, null, null,
          null);
  rmService.init(new Configuration());
  // submit an app and wait for it to block while in app submission
  Thread t = new Thread() {
    @Override
    public void run() {
      try {
        rmService.submitApplication(submitRequest1);
      } catch (YarnException | IOException e) {}
    }
  };
  t.start();
  
  // submit another app, so go through while the first app is blocked
  startBarrier.await();
  rmService.submitApplication(submitRequest2);
  endBarrier.await();
  t.join();
}