EventHandler extends Thread { public void stop(Path intermDir, JobMetadata metadata) { if (inProgressHistFile == null) { LOG.warn("inProgressHistFile is null"); return; } finalHistFile = new Path(intermDir, HistoryFileUtils.generateFileName(metadata)); LOG.info("Stopping event handler thread"); isStopped = true; this.interrupt(); try { this.join(); } catch (InterruptedException e) { LOG.warn("Encountered interruption while stopping event handler thread", e); } cleanUp(); moveInProgressToFinal(); } EventHandler(FileSystem fs, BlockingQueue<Event> q); void setUpThread(Path intermDir, JobMetadata metadata); void emitEvent(Event event); @Override void run(); void stop(Path intermDir, JobMetadata metadata);  }
@Test public void testEventHandlerE2EFailedJobDirNotSet() throws InterruptedException, IOException { fs.mkdirs(jobDir); eventHandlerThread = new EventHandler(fs, eventQueue); eventHandlerThread.start(); eventHandlerThread.stop(null, metadata); eventHandlerThread.join(); assertEquals(fs.listStatus(jobDir).length, 0); Utils.cleanupHDFSPath(fs.getConf(), jobDir); }