@Test public void testMarkOffset() { 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 commitLogRecordDupe = generateRecord(true, false, new OffsetPosition("CommitLog-6-12345.log", 100), new KeyspaceTable("test_keyspace", "test_table")); ChangeRecord commitLogRecordOlderLog = generateRecord(true, false, new OffsetPosition("CommitLog-6-12344.log", 101), 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")); assertFalse(isProcessed(snapshotRecord)); process(snapshotRecord); assertTrue(isProcessed(snapshotRecord)); assertFalse(isProcessed(commitLogRecord)); process(commitLogRecord); assertTrue(isProcessed(commitLogRecord)); assertTrue(isProcessed(commitLogRecordDupe)); process(commitLogRecordDupe); assertTrue(isProcessed(commitLogRecordDupe)); assertTrue(isProcessed(commitLogRecordOlderLog)); process(commitLogRecordOlderLog); assertTrue(isProcessed(commitLogRecordOlderLog)); assertTrue(isProcessed(commitLogRecord)); assertFalse(isProcessed(commitLogRecordDiffTable)); process(commitLogRecordDiffTable); assertTrue(isProcessed(commitLogRecordDiffTable)); }
@Override public void markOffset(String sourceTable, String sourceOffset, boolean isSnapshot) { if (isSnapshot) { synchronized (snapshotOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { snapshotProps.setProperty(sourceTable, sourceOffset); } } } else { synchronized (commitLogOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { commitLogProps.setProperty(sourceTable, sourceOffset); } } } }
FileOffsetWriter implements OffsetWriter { @Override public void markOffset(String sourceTable, String sourceOffset, boolean isSnapshot) { if (isSnapshot) { synchronized (snapshotOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { snapshotProps.setProperty(sourceTable, sourceOffset); } } } else { synchronized (commitLogOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { commitLogProps.setProperty(sourceTable, sourceOffset); } } } } }
FileOffsetWriter implements OffsetWriter { @Override public void markOffset(String sourceTable, String sourceOffset, boolean isSnapshot) { if (isSnapshot) { synchronized (snapshotOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { snapshotProps.setProperty(sourceTable, sourceOffset); } } } else { synchronized (commitLogOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { commitLogProps.setProperty(sourceTable, sourceOffset); } } } } FileOffsetWriter(String offsetDir); }
FileOffsetWriter implements OffsetWriter { @Override public void markOffset(String sourceTable, String sourceOffset, boolean isSnapshot) { if (isSnapshot) { synchronized (snapshotOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { snapshotProps.setProperty(sourceTable, sourceOffset); } } } else { synchronized (commitLogOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { commitLogProps.setProperty(sourceTable, sourceOffset); } } } } 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 markOffset(String sourceTable, String sourceOffset, boolean isSnapshot) { if (isSnapshot) { synchronized (snapshotOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { snapshotProps.setProperty(sourceTable, sourceOffset); } } } else { synchronized (commitLogOffsetFileLock) { if (!isOffsetProcessed(sourceTable, sourceOffset, isSnapshot)) { commitLogProps.setProperty(sourceTable, sourceOffset); } } } } 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; }