@Override
  public void handle(LogHandlerEvent event) {
    switch (event.getType()) {
      case APPLICATION_STARTED:
        LogHandlerAppStartedEvent appStartEvent =
            (LogHandlerAppStartedEvent) event;
        initApp(appStartEvent.getApplicationId(), appStartEvent.getUser(),
            appStartEvent.getCredentials(),
            appStartEvent.getApplicationAcls(),
            appStartEvent.getLogAggregationContext(),
            appStartEvent.getRecoveredAppLogInitedTime());
        break;
      case CONTAINER_FINISHED:
        LogHandlerContainerFinishedEvent containerFinishEvent =
            (LogHandlerContainerFinishedEvent) event;
        stopContainer(containerFinishEvent.getContainerId(),
            containerFinishEvent.getContainerType(),
            containerFinishEvent.getExitCode());
        break;
      case APPLICATION_FINISHED:
        LogHandlerAppFinishedEvent appFinishedEvent =
            (LogHandlerAppFinishedEvent) event;
        stopApp(appFinishedEvent.getApplicationId());
        break;
      case LOG_AGG_TOKEN_UPDATE:
        checkAndEnableAppAggregators();
        break;
      default:
        ; // Ignore
    }

  }
@Test
  public void testAppLogDirCreation() throws Exception {
    final String inputSuffix = "logs-tfile";
    this.conf.set(YarnConfiguration.NM_LOG_DIRS,
        localLogDir.getAbsolutePath());
    this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR,
        this.remoteRootLogDir.getAbsolutePath());
    this.conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX, "logs");

    InlineDispatcher dispatcher = new InlineDispatcher();
    dispatcher.init(this.conf);
    dispatcher.start();

    FileSystem fs = FileSystem.get(this.conf);
    final FileSystem spyFs = spy(FileSystem.get(this.conf));

    final LogAggregationTFileController spyFileFormat
        = new LogAggregationTFileController() {
          @Override
          public FileSystem getFileSystem(Configuration conf)
              throws IOException {
            return spyFs;
          }
        };
    spyFileFormat.initialize(conf, "TFile");
    LogAggregationService aggSvc = new LogAggregationService(dispatcher,
        this.context, this.delSrvc, super.dirsHandler) {
      @Override
      public LogAggregationFileController getLogAggregationFileController(
          Configuration conf) {
        return spyFileFormat;
      }
    };
    aggSvc.init(this.conf);
    aggSvc.start();

    // start an application and verify user, suffix, and app dirs created
    ApplicationId appId = BuilderUtils.newApplicationId(1, 1);
    Path userDir = fs.makeQualified(new Path(
        remoteRootLogDir.getAbsolutePath(), this.user));
    Path bucketDir = fs.makeQualified(LogAggregationUtils.getRemoteBucketDir(
        new Path(remoteRootLogDir.getAbsolutePath()),
            this.user, inputSuffix, appId));
    Path suffixDir = bucketDir.getParent();
    Path appDir = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir(
        new Path(remoteRootLogDir.getAbsolutePath()), appId,
            this.user, inputSuffix));
    LogAggregationContext contextWithAllContainers =
        Records.newRecord(LogAggregationContext.class);
    contextWithAllContainers.setLogAggregationPolicyClassName(
        AllContainerLogAggregationPolicy.class.getName());
    aggSvc.handle(new LogHandlerAppStartedEvent(appId, this.user, null,
        this.acls, contextWithAllContainers));
    verify(spyFs).mkdirs(eq(userDir), isA(FsPermission.class));
    verify(spyFs).mkdirs(eq(suffixDir), isA(FsPermission.class));
    verify(spyFs).mkdirs(eq(bucketDir), isA(FsPermission.class));
    verify(spyFs).mkdirs(eq(appDir), isA(FsPermission.class));

    // start another application and verify only app dir created
    ApplicationId appId2 = BuilderUtils.newApplicationId(1, 2);
    Path appDir2 = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir(
        new Path(remoteRootLogDir.getAbsolutePath()),
            appId2, this.user, inputSuffix));
    aggSvc.handle(new LogHandlerAppStartedEvent(appId2, this.user, null,
        this.acls, contextWithAllContainers));
    verify(spyFs).mkdirs(eq(appDir2), isA(FsPermission.class));

    // start another application with the app dir already created and verify
    // we do not try to create it again
    ApplicationId appId3 = BuilderUtils.newApplicationId(2, 2);
    Path appDir3 = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir(
        new Path(remoteRootLogDir.getAbsolutePath()),
            appId3, this.user, inputSuffix));
    new File(appDir3.toUri().getPath()).mkdir();
    aggSvc.handle(new LogHandlerAppStartedEvent(appId3, this.user, null,
        this.acls, contextWithAllContainers));
    verify(spyFs, never()).mkdirs(eq(appDir3), isA(FsPermission.class));


    // Verify we do not create bucket dir again
    ApplicationId appId4 = BuilderUtils.newApplicationId(2, 10003);
    Path appDir4 = fs.makeQualified(LogAggregationUtils.getRemoteAppLogDir(
            new Path(remoteRootLogDir.getAbsolutePath()),
            appId4, this.user, inputSuffix));
    Path bucketDir4 = appDir4.getParent();
    new File(bucketDir4.toUri().getPath()).mkdir();

    aggSvc.handle(new LogHandlerAppStartedEvent(appId4, this.user, null,
        this.acls, contextWithAllContainers));
    verify(spyFs, never()).mkdirs(eq(bucketDir4), isA(FsPermission.class));
    verify(spyFs).mkdirs(eq(appDir4), isA(FsPermission.class));

    aggSvc.stop();
    aggSvc.close();
    dispatcher.stop();
  }