ignore duplicate events (#133)

* ignore duplicate events

* use configured Comparator to comapre history record

Author: seanzhougoogle

delta-plugins-common/src/main/java/io/cdap/delta/plugin/common/DBSchemaHistory.java

@@ -18,11 +18,13 @@
 
 import io.cdap.cdap.api.common.Bytes;
 import io.cdap.delta.api.DeltaRuntimeContext;
+import io.debezium.config.Configuration;
 import io.debezium.connector.AbstractSourceInfo;
 import io.debezium.document.DocumentReader;
 import io.debezium.document.DocumentWriter;
 import io.debezium.relational.history.AbstractDatabaseHistory;
 import io.debezium.relational.history.DatabaseHistoryException;
+import io.debezium.relational.history.DatabaseHistoryListener;
 import io.debezium.relational.history.HistoryRecord;
 import io.debezium.relational.history.HistoryRecordComparator;
 
@@ -41,11 +43,19 @@ public class DBSchemaHistory extends AbstractDatabaseHistory {
   public static DeltaRuntimeContext deltaRuntimeContext;
   private final DocumentWriter writer = DocumentWriter.defaultWriter();
   private final DocumentReader reader = DocumentReader.defaultReader();
+  private HistoryRecordComparator comparator;
 
   public static void wipeHistory() throws IOException {
     deltaRuntimeContext.putState(KEY, new byte[] { });
   }
 
+  @Override
+  public void configure(Configuration config, HistoryRecordComparator comparator, DatabaseHistoryListener listener,
+    boolean useCatalogBeforeSchema) {
+    this.comparator = comparator == null ? HistoryRecordComparator.INSTANCE : comparator;
+    super.configure(config, comparator, listener, useCatalogBeforeSchema);
+  }
+
   @Override
   protected synchronized void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
     List<HistoryRecord> history = getHistory();
@@ -59,7 +69,7 @@ protected synchronized void storeRecord(HistoryRecord record) throws DatabaseHis
     //And all snapshot history record will have same position
     if (Boolean.TRUE != record.document().getDocument(HistoryRecord.Fields.POSITION)
       .getBoolean(AbstractSourceInfo.SNAPSHOT_KEY) && !history.isEmpty() &&
-      HistoryRecordComparator.INSTANCE.isAtOrBefore(record, history.get(history.size() - 1))) {
+      comparator.isAtOrBefore(record, history.get(history.size() - 1))) {
       return;
     }
     history.add(record);

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerEventReader.java

@@ -147,7 +147,7 @@ public void start(Offset offset) {
       LOG.info("creating new EmbeddedEngine...");
       // Create the engine with this configuration ...
       engine = EmbeddedEngine.create()
-        .notifying(new SqlServerRecordConsumer(context, emitter, databaseName, snapshotTables, sourceTableMap))
+        .notifying(new SqlServerRecordConsumer(context, emitter, databaseName, snapshotTables, sourceTableMap, offset))
         .using(debeziumConf)
         .using(new NotifyingCompletionCallback(context))
         .build();

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerOffset.java

@@ -17,6 +17,7 @@
 package io.cdap.delta.sqlserver;
 
 import io.cdap.delta.api.Offset;
+import io.debezium.connector.sqlserver.Lsn;
 import io.debezium.connector.sqlserver.SourceInfo;
 
 import java.util.HashMap;
@@ -79,6 +80,19 @@ Offset getAsOffset() {
     return new Offset(deltaOffset);
   }
 
+  /**
+   * Returns whether the this {@link SqlServerOffset SqlServerOffset} instance is before or at the specified delta
+   * offset. If it's true that means this {@link SqlServerOffset SqlServerOffset} was once seen by the SQLServer
+   * Debezium connector at the specified delta offset
+   * @param deltaOffset the delta offset to compare
+   * @return whether the this {@link SqlServerOffset SqlServerOffset} instance is before or at the specified delta
+   * offset.
+   */
+  public boolean isBeforeOrAt(Offset deltaOffset) {
+    return Lsn.valueOf(this.changeLsn)
+      .compareTo(Lsn.valueOf(deltaOffset.get().get(SourceInfo.CHANGE_LSN_KEY))) < 1;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerRecordConsumer.java

@@ -53,14 +53,18 @@ public class SqlServerRecordConsumer implements Consumer<SourceRecord> {
   private final String databaseName;
   private final Set<String> snapshotTables;
   private final Map<String, SourceTable> sourceTableMap;
+  private Offset latestOffset;
+
 
   SqlServerRecordConsumer(DeltaSourceContext context, EventEmitter emitter, String databaseName,
-                          Set<String> snapshotTables, Map<String, SourceTable> sourceTableMap) {
+                          Set<String> snapshotTables, Map<String, SourceTable> sourceTableMap,
+    Offset latestOffset) {
     this.context = context;
     this.emitter = emitter;
     this.databaseName = databaseName;
     this.snapshotTables = snapshotTables;
     this.sourceTableMap = sourceTableMap;
+    this.latestOffset = latestOffset;
   }
 
   @Override
@@ -75,9 +79,19 @@ public void accept(SourceRecord sourceRecord) {
     }
 
     SqlServerOffset sqlServerOffset = new SqlServerOffset(sourceRecord.sourceOffset());
+    // ignore duplicated CDC event
+    // SQLServer connector will relay the last event at the offset
+    // to be safe here we check whether it's before or at the same offset
+    // snapshotting will resume from beginning, and the whole table that is partly snapshotted
+    // is supposed to be dropped first , thus no need to consider
+    if (!sqlServerOffset.isSnapshot() && sqlServerOffset.isBeforeOrAt(latestOffset)) {
+      LOG.debug("Got duplicated event {} ", sourceRecord);
+      return;
+    }
+
     sqlServerOffset.setSnapshotTables(snapshotTables);
     boolean isSnapshot = sqlServerOffset.isSnapshot();
-    Offset recordOffset = sqlServerOffset.getAsOffset();
+    latestOffset = sqlServerOffset.getAsOffset();
 
     StructuredRecord val = Records.convert((Struct) sourceRecord.value());
     DMLOperation.Type op;
@@ -143,7 +157,7 @@ public void accept(SourceRecord sourceRecord) {
         primaryFields = fields.stream().map(Schema.Field::getName).collect(Collectors.toList());
       }
       sqlServerOffset.addSnapshotTable(sourceTableId);
-      recordOffset = sqlServerOffset.getAsOffset();
+      Offset recordOffset = sqlServerOffset.getAsOffset();
       builder.setOffset(recordOffset);
 
       try {
@@ -174,7 +188,7 @@ public void accept(SourceRecord sourceRecord) {
 
     Long ingestTime = val.get("ts_ms");
     DMLEvent.Builder dmlBuilder = DMLEvent.builder()
-      .setOffset(recordOffset)
+      .setOffset(latestOffset)
       .setOperationType(op)
       .setDatabaseName(databaseName)
       .setTableName(tableName)