### Question:
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; }### Answer:
@Test public void testProcessChangeRecords() 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 ChangeRecord(sourceInfo, new RowData(), Schema.INT32_SCHEMA, Schema.INT32_SCHEMA, Record.Operation.INSERT, false); queue.enqueue(record); } assertEquals(recordSize, queue.totalCapacity() - queue.remainingCapacity()); queueProcessor.process(); verify(emitter, times(recordSize)).emit(any()); assertEquals(queue.totalCapacity(), queue.remainingCapacity()); }
@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()); }
@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()); }