@Test public void testProcessTombstoneRecords() throws Exception { doNothing().when(emitter).emit(any()); int recordSize = 5; ChangeEventQueue<Event> queue = context.getQueue(); for (int i = 0; i < recordSize; i++) { CassandraConnectorConfig config = new CassandraConnectorConfig(Configuration.from(new Properties())); SourceInfo sourceInfo = new SourceInfo(config, DatabaseDescriptor.getClusterName(), new OffsetPosition("CommitLog-6-123.log", i), new KeyspaceTable(TEST_KEYSPACE, "cdc_table"), false, Conversions.toInstantFromMicros(System.currentTimeMillis() * 1000)); Record record = new TombstoneRecord(sourceInfo, new RowData(), Schema.INT32_SCHEMA); queue.enqueue(record); } assertEquals(recordSize, queue.totalCapacity() - queue.remainingCapacity()); queueProcessor.process(); verify(emitter, times(recordSize)).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; }