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);  }
@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]); } }
@Test public void testHandlingApplicationFinishedEvent() throws IOException { DeletionService delService = new DeletionService(null); NonAggregatingLogHandler aggregatingLogHandler = new NonAggregatingLogHandler(new InlineDispatcher(), delService, dirsHandler); dirsHandler.init(conf); dirsHandler.start(); delService.init(conf); delService.start(); aggregatingLogHandler.init(conf); aggregatingLogHandler.start(); aggregatingLogHandler.handle(new LogHandlerAppFinishedEvent(appId)); aggregatingLogHandler.stop(); aggregatingLogHandler.handle(new LogHandlerAppFinishedEvent(appId)); aggregatingLogHandler.close(); }
@Test @SuppressWarnings("unchecked") public void testLogDeletion() { DeletionService delService = mock(DeletionService.class); Configuration conf = new YarnConfiguration(); String user = "testuser"; File[] localLogDirs = new File[2]; localLogDirs[0] = new File("target", this.getClass().getName() + "-localLogDir0") .getAbsoluteFile(); localLogDirs[1] = new File("target", this.getClass().getName() + "-localLogDir1") .getAbsoluteFile(); 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, 0l); DrainDispatcher dispatcher = createDispatcher(conf); EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class); dispatcher.register(ApplicationEventType.class, appEventHandler); LocalDirsHandlerService dirsHandler = new LocalDirsHandlerService(); dirsHandler.init(conf); ApplicationId appId1 = BuilderUtils.newApplicationId(1234, 1); ApplicationAttemptId appAttemptId1 = BuilderUtils.newApplicationAttemptId(appId1, 1); ContainerId container11 = BuilderUtils.newContainerId(appAttemptId1, 1); NonAggregatingLogHandler logHandler = new NonAggregatingLogHandler(dispatcher, delService, dirsHandler); logHandler.init(conf); logHandler.start(); logHandler.handle(new LogHandlerAppStartedEvent(appId1, user, null, ContainerLogsRetentionPolicy.ALL_CONTAINERS, null)); logHandler.handle(new LogHandlerContainerFinishedEvent(container11, 0)); logHandler.handle(new LogHandlerAppFinishedEvent(appId1)); Path[] localAppLogDirs = new Path[2]; localAppLogDirs[0] = new Path(localLogDirs[0].getAbsolutePath(), appId1.toString()); localAppLogDirs[1] = new Path(localLogDirs[1].getAbsolutePath(), appId1.toString()); long verifyStartTime = System.currentTimeMillis(); WantedButNotInvoked notInvokedException = null; boolean matched = false; while (!matched && System.currentTimeMillis() < verifyStartTime + 5000l) { try { verify(delService).delete(eq(user), (Path) eq(null), eq(localAppLogDirs[0]), eq(localAppLogDirs[1])); matched = true; } catch (WantedButNotInvoked e) { notInvokedException = e; try { Thread.sleep(50l); } catch (InterruptedException i) { } } } if (!matched) { throw notInvokedException; } }
@Test @SuppressWarnings("unchecked") public void testDelayedDelete() { DeletionService delService = mock(DeletionService.class); Configuration conf = new YarnConfiguration(); String user = "testuser"; File[] localLogDirs = new File[2]; localLogDirs[0] = new File("target", this.getClass().getName() + "-localLogDir0") .getAbsoluteFile(); localLogDirs[1] = new File("target", this.getClass().getName() + "-localLogDir1") .getAbsoluteFile(); 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); DrainDispatcher dispatcher = createDispatcher(conf); EventHandler<ApplicationEvent> appEventHandler = mock(EventHandler.class); dispatcher.register(ApplicationEventType.class, appEventHandler); LocalDirsHandlerService dirsHandler = new LocalDirsHandlerService(); dirsHandler.init(conf); ApplicationId appId1 = BuilderUtils.newApplicationId(1234, 1); ApplicationAttemptId appAttemptId1 = BuilderUtils.newApplicationAttemptId(appId1, 1); ContainerId container11 = BuilderUtils.newContainerId(appAttemptId1, 1); NonAggregatingLogHandler logHandler = new NonAggregatingLogHandlerWithMockExecutor(dispatcher, delService, dirsHandler); logHandler.init(conf); logHandler.start(); logHandler.handle(new LogHandlerAppStartedEvent(appId1, user, null, ContainerLogsRetentionPolicy.ALL_CONTAINERS, null)); logHandler.handle(new LogHandlerContainerFinishedEvent(container11, 0)); logHandler.handle(new LogHandlerAppFinishedEvent(appId1)); Path[] localAppLogDirs = new Path[2]; localAppLogDirs[0] = new Path(localLogDirs[0].getAbsolutePath(), appId1.toString()); localAppLogDirs[1] = new Path(localLogDirs[1].getAbsolutePath(), appId1.toString()); ScheduledThreadPoolExecutor mockSched = ((NonAggregatingLogHandlerWithMockExecutor) logHandler).mockSched; verify(mockSched).schedule(any(Runnable.class), eq(10800l), eq(TimeUnit.SECONDS)); }
@Test public void testHandlingApplicationFinishedEvent() { Configuration conf = new Configuration(); LocalDirsHandlerService dirsService = new LocalDirsHandlerService(); DeletionService delService = new DeletionService(null); NonAggregatingLogHandler aggregatingLogHandler = new NonAggregatingLogHandler(new InlineDispatcher(), delService, dirsService); dirsService.init(conf); dirsService.start(); delService.init(conf); delService.start(); aggregatingLogHandler.init(conf); aggregatingLogHandler.start(); ApplicationId appId = BuilderUtils.newApplicationId(1234, 1); aggregatingLogHandler.handle(new LogHandlerAppFinishedEvent(appId)); aggregatingLogHandler.stop(); aggregatingLogHandler.handle(new LogHandlerAppFinishedEvent(appId)); }