@SuppressWarnings("unchecked")
  @Override
  public void handle(LogHandlerEvent event) {
    switch (event.getType()) {
      case APPLICATION_STARTED:
        LogHandlerAppStartedEvent appStartedEvent =
            (LogHandlerAppStartedEvent) event;
        this.appOwners.put(appStartedEvent.getApplicationId(),
            appStartedEvent.getUser());
        this.dispatcher.getEventHandler().handle(
            new ApplicationEvent(appStartedEvent.getApplicationId(),
                ApplicationEventType.APPLICATION_LOG_HANDLING_INITED));
        break;
      case CONTAINER_FINISHED:
        // Ignore
        break;
      case APPLICATION_FINISHED:
        LogHandlerAppFinishedEvent appFinishedEvent =
            (LogHandlerAppFinishedEvent) event;
        ApplicationId appId = appFinishedEvent.getApplicationId();
        String user = appOwners.remove(appId);
        if (user == null) {
          LOG.error("Unable to locate user for {}", appId);
          // send LOG_HANDLING_FAILED out
          NonAggregatingLogHandler.this.dispatcher.getEventHandler().handle(
              new ApplicationEvent(appId,
                  ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED));
          break;
        }
        LogDeleterRunnable logDeleter = new LogDeleterRunnable(user, appId);
        long deletionTimestamp = System.currentTimeMillis()
            + this.deleteDelaySeconds * 1000;
        LogDeleterProto deleterProto = LogDeleterProto.newBuilder()
            .setUser(user)
            .setDeletionTime(deletionTimestamp)
            .build();
        try {
          stateStore.storeLogDeleter(appId, deleterProto);
        } catch (IOException e) {
          LOG.error("Unable to record log deleter state", e);
        }
        try {
          boolean logDeleterStarted = false;
          if (enableTriggerDeleteBySize) {
            final long appLogSize = calculateSizeOfAppLogs(user, appId);
            if (appLogSize >= deleteThreshold) {
              LOG.info("Log Deletion for application: {}, with no delay, size={}", appId, appLogSize);
              sched.schedule(logDeleter, 0, TimeUnit.SECONDS);
              logDeleterStarted = true;
            }
          }
          if (!logDeleterStarted) {
            LOG.info("Scheduling Log Deletion for application: {}, with delay of {} seconds",
                appId, this.deleteDelaySeconds);
            sched.schedule(logDeleter, this.deleteDelaySeconds, TimeUnit.SECONDS);
          }
        } catch (RejectedExecutionException e) {
          // Handling this event in local thread before starting threads
          // or after calling sched.shutdownNow().
          logDeleter.run();
        }
        break;
      default:
    }
  }
@Test
  public void testRecovery() throws Exception {
    File[] localLogDirs = getLocalLogDirFiles(this.getClass().getName(), 2);
    String localLogDirsString =
        localLogDirs[0].getAbsolutePath() + ","
            + localLogDirs[1].getAbsolutePath();

    conf.set(YarnConfiguration.NM_LOG_DIRS, localLogDirsString);
    conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, false);

    conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS,
            YarnConfiguration.DEFAULT_NM_LOG_RETAIN_SECONDS);

    dirsHandler.init(conf);

    appEventHandler.resetLogHandlingEvent();
    assertFalse(appEventHandler.receiveLogHandlingFinishEvent());

    NMStateStoreService stateStore = new NMMemoryStateStoreService();
    stateStore.init(conf);
    stateStore.start();
    NonAggregatingLogHandlerWithMockExecutor logHandler =
        new NonAggregatingLogHandlerWithMockExecutor(dispatcher, mockDelService,
                                                     dirsHandler, stateStore);
    logHandler.init(conf);
    logHandler.start();

    logHandler.handle(new LogHandlerAppStartedEvent(appId, user, null, null));
    logHandler.handle(new LogHandlerContainerFinishedEvent(container11,
        ContainerType.APPLICATION_MASTER, 0));
    logHandler.handle(new LogHandlerAppFinishedEvent(appId));

    // simulate a restart and verify deletion is rescheduled
    logHandler.close();
    logHandler = new NonAggregatingLogHandlerWithMockExecutor(dispatcher,
        mockDelService, dirsHandler, stateStore);
    logHandler.init(conf);
    logHandler.start();
    ArgumentCaptor<Runnable> schedArg = ArgumentCaptor.forClass(Runnable.class);
    verify(logHandler.mockSched).schedule(schedArg.capture(),
        anyLong(), eq(TimeUnit.MILLISECONDS));

    // execute the runnable and verify another restart has nothing scheduled
    schedArg.getValue().run();
    logHandler.close();
    logHandler = new NonAggregatingLogHandlerWithMockExecutor(dispatcher,
        mockDelService, dirsHandler, stateStore);
    logHandler.init(conf);
    logHandler.start();
    verify(logHandler.mockSched, never()).schedule(any(Runnable.class),
        anyLong(), any(TimeUnit.class));

    // wait events get drained.
    this.dispatcher.await();
    assertTrue(appEventHandler.receiveLogHandlingFinishEvent());

    appEventHandler.resetLogHandlingEvent();
    assertFalse(appEventHandler.receiveLogHandlingFailedEvent());
    // send an app finish event against a removed app
    logHandler.handle(new LogHandlerAppFinishedEvent(appId));
    this.dispatcher.await();
    // verify to receive a log failed event.
    assertTrue(appEventHandler.receiveLogHandlingFailedEvent());
    assertFalse(appEventHandler.receiveLogHandlingFinishEvent());
    logHandler.close();
  }