@Test public void testProcessEofEvent() throws Exception { doNothing().when(emitter).emit(any()); ChangeEventQueue<Event> queue = context.getQueue(); File commitLogFile = generateCommitLogFile(); queue.enqueue(new EOFEvent(commitLogFile, true)); assertEquals(1, queue.totalCapacity() - queue.remainingCapacity()); queueProcessor.process(); verify(emitter, times(0)).emit(any()); assertEquals(queue.totalCapacity(), queue.remainingCapacity()); }
@Override public void process() throws InterruptedException { List<Event> events = queue.poll(); for (Event event : events) { processEvent(event); } }
QueueProcessor extends AbstractProcessor { @Override public void process() throws InterruptedException { List<Event> events = queue.poll(); for (Event event : events) { processEvent(event); } } }
QueueProcessor extends AbstractProcessor { @Override public void process() throws InterruptedException { List<Event> events = queue.poll(); for (Event event : events) { processEvent(event); } } QueueProcessor(CassandraConnectorContext context); @VisibleForTesting  QueueProcessor(CassandraConnectorContext context, KafkaRecordEmitter emitter); }
QueueProcessor extends AbstractProcessor { @Override public void process() throws InterruptedException { List<Event> events = queue.poll(); for (Event event : events) { processEvent(event); } } QueueProcessor(CassandraConnectorContext context); @VisibleForTesting  QueueProcessor(CassandraConnectorContext context, KafkaRecordEmitter emitter); @Override void process(); @Override void initialize(); @Override void destroy(); }
QueueProcessor extends AbstractProcessor { @Override public void process() throws InterruptedException { List<Event> events = queue.poll(); for (Event event : events) { processEvent(event); } } QueueProcessor(CassandraConnectorContext context); @VisibleForTesting  QueueProcessor(CassandraConnectorContext context, KafkaRecordEmitter emitter); @Override void process(); @Override void initialize(); @Override void destroy(); static final String ARCHIVE_FOLDER; static final String ERROR_FOLDER; }