@Test public void testFlush() throws IOException { offsetWriter.flush(); try (FileInputStream fis = new FileInputStream(offsetDir.toString() + "/" + FileOffsetWriter.SNAPSHOT_OFFSET_FILE)) { snapshotProps.load(fis); } try (FileInputStream fis = new FileInputStream(offsetDir.toString() + "/" + FileOffsetWriter.COMMITLOG_OFFSET_FILE)) { commitLogProps.load(fis); } assertEquals(0, snapshotProps.size()); assertEquals(0, commitLogProps.size()); ChangeRecord snapshotRecord = generateRecord(true, true, new OffsetPosition("", -1), new KeyspaceTable("test_keyspace", "test_table")); ChangeRecord commitLogRecord = generateRecord(true, false, new OffsetPosition("CommitLog-6-12345.log", 100), new KeyspaceTable("test_keyspace", "test_table")); ChangeRecord commitLogRecordDiffTable = generateRecord(true, false, new OffsetPosition("CommitLog-6-12345.log", 100), new KeyspaceTable("test_keyspace", "test_another_table")); process(snapshotRecord); process(commitLogRecord); process(commitLogRecordDiffTable); offsetWriter.flush(); try (FileInputStream fis = new FileInputStream(offsetDir.toString() + "/" + FileOffsetWriter.SNAPSHOT_OFFSET_FILE)) { snapshotProps.load(fis); } try (FileInputStream fis = new FileInputStream(offsetDir.toString() + "/" + FileOffsetWriter.COMMITLOG_OFFSET_FILE)) { commitLogProps.load(fis); } assertEquals(1, snapshotProps.size()); assertEquals(2, commitLogProps.size()); assertEquals(OffsetPosition.defaultOffsetPosition().serialize(), snapshotProps.getProperty(new KeyspaceTable("test_keyspace", "test_table").name())); assertEquals(new OffsetPosition("CommitLog-6-12345.log", 100).serialize(), commitLogProps.getProperty(new KeyspaceTable("test_keyspace", "test_table").name())); assertEquals(new OffsetPosition("CommitLog-6-12345.log", 100).serialize(), commitLogProps.getProperty(new KeyspaceTable("test_keyspace", "test_another_table").name())); }
@Override public void flush() { try { synchronized (snapshotOffsetFileLock) { saveOffset(snapshotOffsetFile, snapshotProps); } synchronized (commitLogOffsetFileLock) { saveOffset(commitLogOffsetFile, commitLogProps); } } catch (IOException e) { LOGGER.error("Ignoring flush failure", e); } }
FileOffsetWriter implements OffsetWriter { @Override public void flush() { try { synchronized (snapshotOffsetFileLock) { saveOffset(snapshotOffsetFile, snapshotProps); } synchronized (commitLogOffsetFileLock) { saveOffset(commitLogOffsetFile, commitLogProps); } } catch (IOException e) { LOGGER.error("Ignoring flush failure", e); } } }
FileOffsetWriter implements OffsetWriter { @Override public void flush() { try { synchronized (snapshotOffsetFileLock) { saveOffset(snapshotOffsetFile, snapshotProps); } synchronized (commitLogOffsetFileLock) { saveOffset(commitLogOffsetFile, commitLogProps); } } catch (IOException e) { LOGGER.error("Ignoring flush failure", e); } } FileOffsetWriter(String offsetDir); }
FileOffsetWriter implements OffsetWriter { @Override public void flush() { try { synchronized (snapshotOffsetFileLock) { saveOffset(snapshotOffsetFile, snapshotProps); } synchronized (commitLogOffsetFileLock) { saveOffset(commitLogOffsetFile, commitLogProps); } } catch (IOException e) { LOGGER.error("Ignoring flush failure", e); } } FileOffsetWriter(String offsetDir); @Override void markOffset(String sourceTable, String sourceOffset, boolean isSnapshot); @Override boolean isOffsetProcessed(String sourceTable, String sourceOffset, boolean isSnapshot); @Override void flush(); void close(); }
FileOffsetWriter implements OffsetWriter { @Override public void flush() { try { synchronized (snapshotOffsetFileLock) { saveOffset(snapshotOffsetFile, snapshotProps); } synchronized (commitLogOffsetFileLock) { saveOffset(commitLogOffsetFile, commitLogProps); } } catch (IOException e) { LOGGER.error("Ignoring flush failure", e); } } FileOffsetWriter(String offsetDir); @Override void markOffset(String sourceTable, String sourceOffset, boolean isSnapshot); @Override boolean isOffsetProcessed(String sourceTable, String sourceOffset, boolean isSnapshot); @Override void flush(); void close(); static final String SNAPSHOT_OFFSET_FILE; static final String COMMITLOG_OFFSET_FILE; }