@Test public void testSnapshotSkipsNonCdcEnabledTable() 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("non_cdc_table") + " (a int, b text, PRIMARY KEY(a)) WITH cdc = false;"); context.getSchemaHolder().refreshSchemas(); for (int i = 0; i < tableSize; i++) { context.getCassandraClient().execute("INSERT INTO " + keyspaceTable("non_cdc_table") + "(a, b) VALUES (?, ?)", i, String.valueOf(i)); } ChangeEventQueue<Event> queue = context.getQueue(); assertEquals(queue.totalCapacity(), queue.remainingCapacity()); snapshotProcessor.process(); assertEquals(queue.totalCapacity(), queue.remainingCapacity()); 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();  }