@Test public void testSnapshotTable() throws Exception { CassandraConnectorContext context = generateTaskContext(); SnapshotProcessor snapshotProcessor = Mockito.spy(new SnapshotProcessor(context)); when(snapshotProcessor.isRunning()).thenReturn(true); int tableSize = 5; context.getCassandraClient().execute("CREATE TABLE IF NOT EXISTS " + keyspaceTable("cdc_table") + " (a int, b text, PRIMARY KEY(a)) WITH cdc = true;"); context.getCassandraClient().execute("CREATE TABLE IF NOT EXISTS " + keyspaceTable("cdc_table2") + " (a int, b text, PRIMARY KEY(a)) WITH cdc = true;"); context.getSchemaHolder().refreshSchemas(); for (int i = 0; i < tableSize; i++) { context.getCassandraClient().execute("INSERT INTO " + keyspaceTable("cdc_table") + "(a, b) VALUES (?, ?)", i, String.valueOf(i)); context.getCassandraClient().execute("INSERT INTO " + keyspaceTable("cdc_table2") + "(a, b) VALUES (?, ?)", i + 10, String.valueOf(i + 10)); } ChangeEventQueue<Event> queue = context.getQueue(); assertEquals(queue.totalCapacity(), queue.remainingCapacity()); snapshotProcessor.process(); assertEquals(2 * tableSize, queue.totalCapacity() - queue.remainingCapacity()); final List<ChangeRecord> table1 = new ArrayList<>(); final List<ChangeRecord> table2 = new ArrayList<>(); for (Event event : queue.poll()) { ChangeRecord record = (ChangeRecord) event; Assert.assertEquals(record.getEventType(), Event.EventType.CHANGE_EVENT); Assert.assertEquals(record.getOp(), Record.Operation.INSERT); assertEquals(record.getSource().cluster, DatabaseDescriptor.getClusterName()); assertTrue(record.getSource().snapshot); final String tableName = record.getSource().keyspaceTable.name(); if (tableName.equals(keyspaceTable("cdc_table"))) { table1.add(record); } else { table2.add(record); } Assert.assertEquals(record.getSource().offsetPosition, OffsetPosition.defaultOffsetPosition()); } assertEquals(tableSize, table1.size()); assertEquals(tableSize, table2.size()); deleteTestKeyspaceTables(); deleteTestOffsets(context); context.cleanUp(); }
@Override public void process() { if (snapshotMode == CassandraConnectorConfig.SnapshotMode.ALWAYS) { snapshot(); } else if (snapshotMode == CassandraConnectorConfig.SnapshotMode.INITIAL && initial) { snapshot(); initial = false; } else { LOGGER.debug("Skipping snapshot [mode: {}]", snapshotMode); } }
SnapshotProcessor extends AbstractProcessor { @Override public void process() { if (snapshotMode == CassandraConnectorConfig.SnapshotMode.ALWAYS) { snapshot(); } else if (snapshotMode == CassandraConnectorConfig.SnapshotMode.INITIAL && initial) { snapshot(); initial = false; } else { LOGGER.debug("Skipping snapshot [mode: {}]", snapshotMode); } } }
SnapshotProcessor extends AbstractProcessor { @Override public void process() { if (snapshotMode == CassandraConnectorConfig.SnapshotMode.ALWAYS) { snapshot(); } else if (snapshotMode == CassandraConnectorConfig.SnapshotMode.INITIAL && initial) { snapshot(); initial = false; } else { LOGGER.debug("Skipping snapshot [mode: {}]", snapshotMode); } } SnapshotProcessor(CassandraConnectorContext context); }
SnapshotProcessor extends AbstractProcessor { @Override public void process() { if (snapshotMode == CassandraConnectorConfig.SnapshotMode.ALWAYS) { snapshot(); } else if (snapshotMode == CassandraConnectorConfig.SnapshotMode.INITIAL && initial) { snapshot(); initial = false; } else { LOGGER.debug("Skipping snapshot [mode: {}]", snapshotMode); } } SnapshotProcessor(CassandraConnectorContext context); @Override void initialize(); @Override void destroy(); @Override void process(); }
SnapshotProcessor extends AbstractProcessor { @Override public void process() { if (snapshotMode == CassandraConnectorConfig.SnapshotMode.ALWAYS) { snapshot(); } else if (snapshotMode == CassandraConnectorConfig.SnapshotMode.INITIAL && initial) { snapshot(); initial = false; } else { LOGGER.debug("Skipping snapshot [mode: {}]", snapshotMode); } } SnapshotProcessor(CassandraConnectorContext context); @Override void initialize(); @Override void destroy(); @Override void process();  }