not store duplicate history schema record

clear schema history if replicator was stopped in middle of snapshot

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.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.HistoryRecord;
+import io.debezium.relational.history.HistoryRecordComparator;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -47,6 +49,19 @@ public static void wipeHistory() throws IOException {
   @Override
   protected synchronized void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
     List<HistoryRecord> history = getHistory();
+    //ignore the history record already seen
+    //we serialize the history record once DDL event is seen in the source
+    //however offset is committed once event is applied in the target
+    //so it's possible that debezium is resuming from a point that is earlier than the last
+    //serialized history record. And when recover the history record, debezium will ignore those
+    //history record that is later than the resuming point.
+    //Thus it's possible for Debezium to store some history record that is already serialized.
+    //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))) {
+      return;
+    }
     history.add(record);
     String historyStr = history.stream().map(r -> {
       try {

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlEventReader.java

@@ -17,7 +17,6 @@
 package io.cdap.delta.mysql;
 
 import com.google.common.annotations.VisibleForTesting;
-import io.cdap.delta.api.DeltaFailureException;
 import io.cdap.delta.api.DeltaSourceContext;
 import io.cdap.delta.api.EventEmitter;
 import io.cdap.delta.api.EventReader;
@@ -93,14 +92,15 @@ public void start(Offset offset) {
     Map<String, SourceTable> sourceTableMap = sourceTables.stream().collect(
       Collectors.toMap(t -> config.getDatabase() + "." + t.getTable(), t -> t));
     Map<String, String> state = offset.get(); // state map is always not null
+    String isSnapshot = state.getOrDefault(MySqlConstantOffsetBackingStore.SNAPSHOT, "");
     Configuration.Builder configBuilder = Configuration.create()
       .with("connector.class", MySqlConnector.class.getName())
       .with("offset.storage", MySqlConstantOffsetBackingStore.class.getName())
       .with("offset.flush.interval.ms", 1000)
       /* bind offset configs with debeizumConf */
       .with("file", state.getOrDefault(MySqlConstantOffsetBackingStore.FILE, ""))
       .with("pos", state.getOrDefault(MySqlConstantOffsetBackingStore.POS, ""))
-      .with("snapshot", state.getOrDefault(MySqlConstantOffsetBackingStore.SNAPSHOT, ""))
+      .with("snapshot", isSnapshot)
       .with("row", state.getOrDefault(MySqlConstantOffsetBackingStore.ROW, ""))
       .with("event", state.getOrDefault(MySqlConstantOffsetBackingStore.EVENT, ""))
       .with("gtids", state.getOrDefault(MySqlConstantOffsetBackingStore.GTID_SET, ""))
@@ -125,13 +125,11 @@ public void start(Offset offset) {
     MySqlConnectorConfig mysqlConf = new MySqlConnectorConfig(debeziumConf);
     DBSchemaHistory.deltaRuntimeContext = context;
     /*
-       this is required in scenarios where the source is able to emit the starting DDL events during snapshotting,
-       but the target is unable to apply them. In that case, this reader will be created again, but it won't re-emit
-       those DDL events unless the DB history is wiped. This only fixes handling of DDL errors that
-       happen during the initial snapshot.
-        TODO: (CDAP-16735) fix this more comprehensively
+     * All snapshot events or schema history record have same position/offset
+     * if replicator was stopped  or paused from middle of snapshot, it
+     * will resume from beginning.
      */
-    if (offset.get().isEmpty()) {
+    if (offset.get().isEmpty() || "true".equalsIgnoreCase(isSnapshot)) {
       try {
         DBSchemaHistory.wipeHistory();
       } catch (IOException e) {

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlRecordConsumer.java

@@ -126,7 +126,7 @@ public void accept(SourceRecord sourceRecord) {
       // This should not happen, 'source' is a mandatory field in sourceRecord from debezium
       return;
     }
-    boolean isSnapshot = Boolean.TRUE.equals(source.get(MySqlConstantOffsetBackingStore.SNAPSHOT));
+    boolean isSnapshot = Boolean.TRUE.equals(deltaOffset.get(MySqlConstantOffsetBackingStore.SNAPSHOT));
     // If the map is empty, we should read all DDL/DML events and columns of all tables
     boolean readAllTables = sourceTableMap.isEmpty();
 

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

@@ -98,6 +98,7 @@ public void start(Offset offset) {
 
     Map<String, String> state = offset.get(); // this will never be null
     // offset config
+    String isSnapshotCompleted = state.getOrDefault(SqlServerConstantOffsetBackingStore.SNAPSHOT_COMPLETED, "");
     Configuration debeziumConf = Configuration.create()
       .with("connector.class", SqlServerConnector.class.getName())
       .with("offset.storage", SqlServerConstantOffsetBackingStore.class.getName())
@@ -106,7 +107,7 @@ public void start(Offset offset) {
       .with("change_lsn", state.getOrDefault(SourceInfo.CHANGE_LSN_KEY, ""))
       .with("commit_lsn", state.getOrDefault(SourceInfo.COMMIT_LSN_KEY, ""))
       .with("snapshot", state.getOrDefault(SourceInfo.SNAPSHOT_KEY, ""))
-      .with("snapshot_completed", state.getOrDefault(SqlServerConstantOffsetBackingStore.SNAPSHOT_COMPLETED, ""))
+      .with("snapshot_completed", isSnapshotCompleted)
       /* begin connector properties */
       .with("name", "delta")
       .with("database.hostname", config.getHost())
@@ -127,13 +128,11 @@ public void start(Offset offset) {
       new HashSet<>(Arrays.asList(snapshotTablesStr.split(SqlServerOffset.DELIMITER)));
 
     /*
-       this is required in scenarios where the source is able to emit the starting DDL events during snapshotting,
-       but the target is unable to apply them. In that case, this reader will be created again, but it won't re-emit
-       those DDL events unless the DB history is wiped. This only fixes handling of DDL errors that
-       happen during the initial snapshot.
-        TODO: (CDAP-16735) fix this more comprehensively
+     * All snapshot events or schema history record have same position/offset
+     * if replicator was stopped  or paused from middle of snapshot, it
+     * will resume from beginning.
      */
-    if (offset.get().isEmpty()) {
+    if (offset.get().isEmpty() || !"true".equalsIgnoreCase(isSnapshotCompleted)) {
       try {
         DBSchemaHistory.wipeHistory();
       } catch (IOException e) {