@Test public void testDelayedDelete() throws IOException { 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); NonAggregatingLogHandler logHandler = new NonAggregatingLogHandlerWithMockExecutor(dispatcher, mockDelService, dirsHandler); logHandler.init(conf); logHandler.start(); logHandler.handle(new LogHandlerAppStartedEvent(appId, user, null, ContainerLogsRetentionPolicy.ALL_CONTAINERS, null)); logHandler.handle(new LogHandlerContainerFinishedEvent(container11, 0)); logHandler.handle(new LogHandlerAppFinishedEvent(appId)); Path[] localAppLogDirs = new Path[2]; localAppLogDirs[0] = new Path(localLogDirs[0].getAbsolutePath(), appId.toString()); localAppLogDirs[1] = new Path(localLogDirs[1].getAbsolutePath(), appId.toString()); ScheduledThreadPoolExecutor mockSched = ((NonAggregatingLogHandlerWithMockExecutor) logHandler).mockSched; verify(mockSched).schedule(any(Runnable.class), eq(10800l), eq(TimeUnit.SECONDS)); logHandler.close(); for (int i = 0; i < localLogDirs.length; i++) { FileUtils.deleteDirectory(localLogDirs[i]); } }
@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: break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; LOG.info("Scheduling Log Deletion for application: " + appFinishedEvent.getApplicationId() + ", with delay of " + this.deleteDelaySeconds + " seconds"); LogDeleterRunnable logDeleter = new LogDeleterRunnable(appOwners.remove(appFinishedEvent .getApplicationId()), appFinishedEvent.getApplicationId()); try { sched.schedule(logDeleter, this.deleteDelaySeconds, TimeUnit.SECONDS); } catch (RejectedExecutionException e) { logDeleter.run(); } break; default: ; } }
NonAggregatingLogHandler extends AbstractService implements
    LogHandler { @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: break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; LOG.info("Scheduling Log Deletion for application: " + appFinishedEvent.getApplicationId() + ", with delay of " + this.deleteDelaySeconds + " seconds"); LogDeleterRunnable logDeleter = new LogDeleterRunnable(appOwners.remove(appFinishedEvent .getApplicationId()), appFinishedEvent.getApplicationId()); try { sched.schedule(logDeleter, this.deleteDelaySeconds, TimeUnit.SECONDS); } catch (RejectedExecutionException e) { logDeleter.run(); } break; default: ; } } }
NonAggregatingLogHandler extends AbstractService implements
    LogHandler { @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: break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; LOG.info("Scheduling Log Deletion for application: " + appFinishedEvent.getApplicationId() + ", with delay of " + this.deleteDelaySeconds + " seconds"); LogDeleterRunnable logDeleter = new LogDeleterRunnable(appOwners.remove(appFinishedEvent .getApplicationId()), appFinishedEvent.getApplicationId()); try { sched.schedule(logDeleter, this.deleteDelaySeconds, TimeUnit.SECONDS); } catch (RejectedExecutionException e) { logDeleter.run(); } break; default: ; } } NonAggregatingLogHandler(Dispatcher dispatcher,
      DeletionService delService, LocalDirsHandlerService dirsHandler); }
NonAggregatingLogHandler extends AbstractService implements
    LogHandler { @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: break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; LOG.info("Scheduling Log Deletion for application: " + appFinishedEvent.getApplicationId() + ", with delay of " + this.deleteDelaySeconds + " seconds"); LogDeleterRunnable logDeleter = new LogDeleterRunnable(appOwners.remove(appFinishedEvent .getApplicationId()), appFinishedEvent.getApplicationId()); try { sched.schedule(logDeleter, this.deleteDelaySeconds, TimeUnit.SECONDS); } catch (RejectedExecutionException e) { logDeleter.run(); } break; default: ; } } NonAggregatingLogHandler(Dispatcher dispatcher,
      DeletionService delService, LocalDirsHandlerService dirsHandler); @SuppressWarnings("unchecked") @Override void handle(LogHandlerEvent event); }
NonAggregatingLogHandler extends AbstractService implements
    LogHandler { @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: break; case APPLICATION_FINISHED: LogHandlerAppFinishedEvent appFinishedEvent = (LogHandlerAppFinishedEvent) event; LOG.info("Scheduling Log Deletion for application: " + appFinishedEvent.getApplicationId() + ", with delay of " + this.deleteDelaySeconds + " seconds"); LogDeleterRunnable logDeleter = new LogDeleterRunnable(appOwners.remove(appFinishedEvent .getApplicationId()), appFinishedEvent.getApplicationId()); try { sched.schedule(logDeleter, this.deleteDelaySeconds, TimeUnit.SECONDS); } catch (RejectedExecutionException e) { logDeleter.run(); } break; default: ; } } NonAggregatingLogHandler(Dispatcher dispatcher,
      DeletionService delService, LocalDirsHandlerService dirsHandler); @SuppressWarnings("unchecked") @Override void handle(LogHandlerEvent event);  }