@Test public void testDrainQueue() { DataFileWriter<Event> writer = mock(DataFileWriter.class); eventQueue.add(eEventWrapper); eventQueue.add(eEventWrapper); eventQueue.add(eEventWrapper); eventQueue.add(eEventWrapper); eventHandlerThread = new EventHandler(fs, eventQueue); assertEquals(eventQueue.size(), 4); eventHandlerThread.drainQueue(eventQueue, writer); assertEquals(eventQueue.size(), 0); }
@VisibleForTesting void drainQueue(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { LOG.info("Draining queue"); while (!eventQueue.isEmpty()) { try { Event event = queue.poll(); writer.append(event); } catch (IOException e) { LOG.error("Failed to drain queue", e); } } }
EventHandler extends Thread { @VisibleForTesting void drainQueue(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { LOG.info("Draining queue"); while (!eventQueue.isEmpty()) { try { Event event = queue.poll(); writer.append(event); } catch (IOException e) { LOG.error("Failed to drain queue", e); } } } }
EventHandler extends Thread { @VisibleForTesting void drainQueue(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { LOG.info("Draining queue"); while (!eventQueue.isEmpty()) { try { Event event = queue.poll(); writer.append(event); } catch (IOException e) { LOG.error("Failed to drain queue", e); } } } EventHandler(FileSystem fs, BlockingQueue<Event> q); }
EventHandler extends Thread { @VisibleForTesting void drainQueue(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { LOG.info("Draining queue"); while (!eventQueue.isEmpty()) { try { Event event = queue.poll(); writer.append(event); } catch (IOException e) { LOG.error("Failed to drain queue", e); } } } 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 drainQueue(BlockingQueue<Event> queue, DataFileWriter<Event> writer) { LOG.info("Draining queue"); while (!eventQueue.isEmpty()) { try { Event event = queue.poll(); writer.append(event); } catch (IOException e) { LOG.error("Failed to drain queue", e); } } } 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);  }