@Test public void testWriteEvent() throws IOException { DataFileWriter<Event> writer = mock(DataFileWriter.class); eventQueue.add(eEventWrapper); eventHandlerThread = new EventHandler(fs, eventQueue); assertEquals(eventQueue.size(), 1); eventHandlerThread.writeEvent(eventQueue, writer); assertEquals(eventQueue.size(), 0); verify(writer).append(eEventWrapper); }
@VisibleForTesting void writeEvent(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { Event event = null; try { event = queue.take(); writer.append(event); } catch (IOException e) { LOG.error("Failed to append event " + event, e); } catch (InterruptedException e) { LOG.info("Event writer interrupted"); } }
EventHandler extends Thread { @VisibleForTesting void writeEvent(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { Event event = null; try { event = queue.take(); writer.append(event); } catch (IOException e) { LOG.error("Failed to append event " + event, e); } catch (InterruptedException e) { LOG.info("Event writer interrupted"); } } }
EventHandler extends Thread { @VisibleForTesting void writeEvent(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { Event event = null; try { event = queue.take(); writer.append(event); } catch (IOException e) { LOG.error("Failed to append event " + event, e); } catch (InterruptedException e) { LOG.info("Event writer interrupted"); } } EventHandler(FileSystem fs, BlockingQueue<Event> q); }
EventHandler extends Thread { @VisibleForTesting void writeEvent(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { Event event = null; try { event = queue.take(); writer.append(event); } catch (IOException e) { LOG.error("Failed to append event " + event, e); } catch (InterruptedException e) { LOG.info("Event writer interrupted"); } } 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); }
EventHandler extends Thread { @VisibleForTesting void writeEvent(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { Event event = null; try { event = queue.take(); writer.append(event); } catch (IOException e) { LOG.error("Failed to append event " + event, e); } catch (InterruptedException e) { LOG.info("Event writer interrupted"); } } 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);  }