@Test public void testProcessCommitLogs() throws Exception { int commitLogRowSize = 10; context.getCassandraClient().execute("CREATE TABLE IF NOT EXISTS " + keyspaceTable("cdc_table") + " (a int, b int, PRIMARY KEY(a)) WITH cdc = true;"); context.getSchemaHolder().refreshSchemas(); CFMetaData cfMetaData = Schema.instance.getCFMetaData(TEST_KEYSPACE, "cdc_table"); for (int i = 0; i < commitLogRowSize; i++) { SimpleBuilders.PartitionUpdateBuilder puBuilder = new SimpleBuilders.PartitionUpdateBuilder(cfMetaData, i); Row row = puBuilder.row().add("b", i).build(); PartitionUpdate pu = PartitionUpdate.singleRowUpdate(cfMetaData, puBuilder.build().partitionKey(), row); Mutation m = new Mutation(pu); CommitLog.instance.add(m); } CommitLog.instance.sync(true); ChangeEventQueue<Event> queue = context.getQueue(); assertEquals(queue.totalCapacity(), queue.remainingCapacity()); File cdcLoc = new File(DatabaseDescriptor.getCommitLogLocation()); File[] commitLogs = CommitLogUtil.getCommitLogs(cdcLoc); for (File commitLog : commitLogs) { commitLogProcessor.processCommitLog(commitLog); } List<Event> events = queue.poll(); int eofEventSize = commitLogs.length; assertEquals(commitLogRowSize + eofEventSize, events.size()); for (int i = 0; i < events.size(); i++) { Event event = events.get(i); if (event instanceof Record) { Record record = (Record) events.get(i); assertEquals(record.getEventType(), Event.EventType.CHANGE_EVENT); assertEquals(record.getSource().cluster, DatabaseDescriptor.getClusterName()); assertFalse(record.getSource().snapshot); assertEquals(record.getSource().keyspaceTable.name(), keyspaceTable("cdc_table")); assertTrue(record.getSource().offsetPosition.fileName.contains(String.valueOf(CommitLog.instance.getCurrentPosition().segmentId))); } else if (event instanceof EOFEvent) { EOFEvent eofEvent = (EOFEvent) event; assertTrue(eofEvent.success); } else { throw new Exception("unexpected event type"); } } deleteTestKeyspaceTables(); }
void processCommitLog(File file) throws IOException { if (file == null) { throw new IOException("Commit log is null"); } if (!file.exists()) { throw new IOException("Commit log " + file.getName() + " does not exist"); } try { try { LOGGER.info("Processing commit log {}", file.getName()); metrics.setCommitLogFilename(file.getName()); commitLogReader.readCommitLogSegment(commitLogReadHandler, file, false); if (!latestOnly) { queue.enqueue(new EOFEvent(file, true)); } LOGGER.info("Successfully processed commit log {}", file.getName()); } catch (Exception e) { if (!latestOnly) { queue.enqueue(new EOFEvent(file, false)); } if (commitLogTransfer.getClass().getName().equals(CassandraConnectorConfig.DEFAULT_COMMIT_LOG_TRANSFER_CLASS)) { LOGGER.error("Error occurred while processing commit log " + file.getName(), e); throw e; } LOGGER.warn("Error occurred while processing commit log " + file.getName(), e); } } catch (InterruptedException e) { LOGGER.error("Interruption while enqueuing EOF Event for file {}", file.getName()); throw new CassandraConnectorTaskException("Enqueuing has been interrupted: ", e); } }
CommitLogProcessor extends AbstractProcessor { void processCommitLog(File file) throws IOException { if (file == null) { throw new IOException("Commit log is null"); } if (!file.exists()) { throw new IOException("Commit log " + file.getName() + " does not exist"); } try { try { LOGGER.info("Processing commit log {}", file.getName()); metrics.setCommitLogFilename(file.getName()); commitLogReader.readCommitLogSegment(commitLogReadHandler, file, false); if (!latestOnly) { queue.enqueue(new EOFEvent(file, true)); } LOGGER.info("Successfully processed commit log {}", file.getName()); } catch (Exception e) { if (!latestOnly) { queue.enqueue(new EOFEvent(file, false)); } if (commitLogTransfer.getClass().getName().equals(CassandraConnectorConfig.DEFAULT_COMMIT_LOG_TRANSFER_CLASS)) { LOGGER.error("Error occurred while processing commit log " + file.getName(), e); throw e; } LOGGER.warn("Error occurred while processing commit log " + file.getName(), e); } } catch (InterruptedException e) { LOGGER.error("Interruption while enqueuing EOF Event for file {}", file.getName()); throw new CassandraConnectorTaskException("Enqueuing has been interrupted: ", e); } } }
CommitLogProcessor extends AbstractProcessor { void processCommitLog(File file) throws IOException { if (file == null) { throw new IOException("Commit log is null"); } if (!file.exists()) { throw new IOException("Commit log " + file.getName() + " does not exist"); } try { try { LOGGER.info("Processing commit log {}", file.getName()); metrics.setCommitLogFilename(file.getName()); commitLogReader.readCommitLogSegment(commitLogReadHandler, file, false); if (!latestOnly) { queue.enqueue(new EOFEvent(file, true)); } LOGGER.info("Successfully processed commit log {}", file.getName()); } catch (Exception e) { if (!latestOnly) { queue.enqueue(new EOFEvent(file, false)); } if (commitLogTransfer.getClass().getName().equals(CassandraConnectorConfig.DEFAULT_COMMIT_LOG_TRANSFER_CLASS)) { LOGGER.error("Error occurred while processing commit log " + file.getName(), e); throw e; } LOGGER.warn("Error occurred while processing commit log " + file.getName(), e); } } catch (InterruptedException e) { LOGGER.error("Interruption while enqueuing EOF Event for file {}", file.getName()); throw new CassandraConnectorTaskException("Enqueuing has been interrupted: ", e); } } CommitLogProcessor(CassandraConnectorContext context); }
CommitLogProcessor extends AbstractProcessor { void processCommitLog(File file) throws IOException { if (file == null) { throw new IOException("Commit log is null"); } if (!file.exists()) { throw new IOException("Commit log " + file.getName() + " does not exist"); } try { try { LOGGER.info("Processing commit log {}", file.getName()); metrics.setCommitLogFilename(file.getName()); commitLogReader.readCommitLogSegment(commitLogReadHandler, file, false); if (!latestOnly) { queue.enqueue(new EOFEvent(file, true)); } LOGGER.info("Successfully processed commit log {}", file.getName()); } catch (Exception e) { if (!latestOnly) { queue.enqueue(new EOFEvent(file, false)); } if (commitLogTransfer.getClass().getName().equals(CassandraConnectorConfig.DEFAULT_COMMIT_LOG_TRANSFER_CLASS)) { LOGGER.error("Error occurred while processing commit log " + file.getName(), e); throw e; } LOGGER.warn("Error occurred while processing commit log " + file.getName(), e); } } catch (InterruptedException e) { LOGGER.error("Interruption while enqueuing EOF Event for file {}", file.getName()); throw new CassandraConnectorTaskException("Enqueuing has been interrupted: ", e); } } CommitLogProcessor(CassandraConnectorContext context); @Override void initialize(); @Override void destroy(); @Override void process(); }
CommitLogProcessor extends AbstractProcessor { void processCommitLog(File file) throws IOException { if (file == null) { throw new IOException("Commit log is null"); } if (!file.exists()) { throw new IOException("Commit log " + file.getName() + " does not exist"); } try { try { LOGGER.info("Processing commit log {}", file.getName()); metrics.setCommitLogFilename(file.getName()); commitLogReader.readCommitLogSegment(commitLogReadHandler, file, false); if (!latestOnly) { queue.enqueue(new EOFEvent(file, true)); } LOGGER.info("Successfully processed commit log {}", file.getName()); } catch (Exception e) { if (!latestOnly) { queue.enqueue(new EOFEvent(file, false)); } if (commitLogTransfer.getClass().getName().equals(CassandraConnectorConfig.DEFAULT_COMMIT_LOG_TRANSFER_CLASS)) { LOGGER.error("Error occurred while processing commit log " + file.getName(), e); throw e; } LOGGER.warn("Error occurred while processing commit log " + file.getName(), e); } } catch (InterruptedException e) { LOGGER.error("Interruption while enqueuing EOF Event for file {}", file.getName()); throw new CassandraConnectorTaskException("Enqueuing has been interrupted: ", e); } } CommitLogProcessor(CassandraConnectorContext context); @Override void initialize(); @Override void destroy(); @Override void process();  }