@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();
}