Merge pull request #127 from data-integrations/fix/schema-history

Fix Schema History

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;
@@ -37,33 +39,30 @@ public class DBSchemaHistory extends AbstractDatabaseHistory {
   private static final String KEY = "history";
   // Hacky, fix when usage of EmbeddedEngine is replaced
   public static DeltaRuntimeContext deltaRuntimeContext;
-  private static final DocumentWriter writer = DocumentWriter.defaultWriter();
-  private static final DocumentReader reader = DocumentReader.defaultReader();
+  private final DocumentWriter writer = DocumentWriter.defaultWriter();
+  private final DocumentReader reader = DocumentReader.defaultReader();
 
-  /**
-   * Restart the DB schema history from certain {@link HistoryRecord HistoryRecord} specified by the parameter
-   * @param index the position of the {@link HistoryRecord HistoryRecord} to restart from in the history list. A
-   *              negative value will wipe out all the history
-   */
-  public static void restartFrom(int index) throws IOException {
-    if (index < 0) {
-      wipeHistory();
-      return;
-    }
-    storeHistory(getHistory().subList(0, index + 1));
-  }
   public static void wipeHistory() throws IOException {
     deltaRuntimeContext.putState(KEY, new byte[] { });
   }
 
   @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);
-    storeHistory(history);
-  }
-
-  private static void storeHistory(List<HistoryRecord> history) {
     String historyStr = history.stream().map(r -> {
       try {
         return writer.write(r.document());
@@ -100,7 +99,7 @@ public boolean storageExists() {
   }
 
   // TODO: cache history, should only have to read once
-  private static List<HistoryRecord> getHistory() {
+  private List<HistoryRecord> getHistory() {
     List<HistoryRecord> history = new ArrayList<>();
     try {
       byte[] historyBytes = deltaRuntimeContext.getState(KEY);

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

@@ -38,7 +38,6 @@
 import io.debezium.jdbc.TemporalPrecisionMode;
 import io.debezium.relational.Tables;
 import io.debezium.relational.ddl.DdlParser;
-import io.debezium.relational.history.HistoryRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,8 +58,6 @@
  */
 public class MySqlEventReader implements EventReader {
   public static final Logger LOG = LoggerFactory.getLogger(MySqlEventReader.class);
-
-  static final String SCHEMA_HISTORY_INDEX = "schema.history";
   private final MySqlConfig config;
   private final EventEmitter emitter;
   private final ExecutorService executorService;
@@ -95,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, ""))
@@ -126,16 +124,17 @@ public void start(Offset offset) {
     Configuration debeziumConf = configBuilder.build();
     MySqlConnectorConfig mysqlConf = new MySqlConnectorConfig(debeziumConf);
     DBSchemaHistory.deltaRuntimeContext = context;
-    /**
-     * DBSchemaHistory stores the historical DDL changes. Each time Debezium detects a DDL change in binlog, it will
-     * store a {@link HistoryRecord History Record} to DBSchemaHistory, in case of a failure or on purpose stop,
-     * It knows where to restart from.
+    /*
+     * 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.
      */
-    int schemaHistoryIndex = Integer.parseInt(state.getOrDefault(SCHEMA_HISTORY_INDEX, "-1"));
-    try {
-      DBSchemaHistory.restartFrom(schemaHistoryIndex);
-    } catch (IOException e) {
-      throw new RuntimeException("Unable to reset the schema history at start of replication.", e);
+    if (offset.get().isEmpty() || "true".equalsIgnoreCase(isSnapshot)) {
+      try {
+        DBSchemaHistory.wipeHistory();
+      } catch (IOException e) {
+        throw new RuntimeException("Unable to wipe schema history at start of replication.", e);
+      }
     }
 
     MySqlValueConverters mySqlValueConverters = getValueConverters(mysqlConf);
@@ -148,7 +147,7 @@ public void start(Offset offset) {
       engine = EmbeddedEngine.create()
         .using(debeziumConf)
         .notifying(new MySqlRecordConsumer(context, emitter, ddlParser, mySqlValueConverters,
-                                           new Tables(), sourceTableMap, schemaHistoryIndex))
+                                           new Tables(), sourceTableMap))
         .using(new NotifyingCompletionCallback(context))
         .build();
       executorService.submit(engine);

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

@@ -56,18 +56,16 @@ public class MySqlRecordConsumer implements Consumer<SourceRecord> {
   private final MySqlValueConverters mySqlValueConverters;
   private final Tables tables;
   private final Map<String, SourceTable> sourceTableMap;
-  private int schemaHistoryIndex;
 
   public MySqlRecordConsumer(DeltaSourceContext context, EventEmitter emitter,
                              DdlParser ddlParser, MySqlValueConverters mySqlValueConverters,
-                             Tables tables, Map<String, SourceTable> sourceTableMap, int schemaHistoryIndex) {
+                             Tables tables, Map<String, SourceTable> sourceTableMap) {
     this.context = context;
     this.emitter = emitter;
     this.ddlParser = ddlParser;
     this.mySqlValueConverters = mySqlValueConverters;
     this.tables = tables;
     this.sourceTableMap = sourceTableMap;
-    this.schemaHistoryIndex = schemaHistoryIndex;
   }
 
   @Override
@@ -118,22 +116,23 @@ public void accept(SourceRecord sourceRecord) {
       return;
     }
 
+    Map<String, String> deltaOffset = generateCdapOffsets(sourceRecord);
+    Offset recordOffset = new Offset(deltaOffset);
 
     StructuredRecord val = Records.convert((Struct) sourceRecord.value());
+    String ddl = val.get("ddl");
     StructuredRecord source = val.get("source");
     if (source == null) {
       // 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();
 
-    String ddl = val.get("ddl");
-    Map<String, String> deltaOffset = generateCdapOffsets(sourceRecord);
     try {
       if (ddl != null) {
-        handleDDL(ddl, deltaOffset, isSnapshot, readAllTables);
+        handleDDL(ddl, recordOffset, isSnapshot, readAllTables);
         return;
       }
 
@@ -144,16 +143,15 @@ public void accept(SourceRecord sourceRecord) {
         return;
       }
 
-      handleDML(source, val, deltaOffset, isSnapshot, readAllTables);
+      handleDML(source, val, recordOffset, isSnapshot, readAllTables);
     } catch (InterruptedException e) {
       // happens when the event reader is stopped. throwing this exception tells Debezium to stop right away
       throw new StopConnectorException("Interrupted while emitting event.");
     }
   }
 
-  private void handleDML(StructuredRecord source, StructuredRecord val, Map<String, String> deltaOffset,
+  private void handleDML(StructuredRecord source, StructuredRecord val, Offset recordOffset,
                          boolean isSnapshot, boolean readAllTables) throws InterruptedException {
-    deltaOffset.put(MySqlEventReader.SCHEMA_HISTORY_INDEX, String.valueOf(schemaHistoryIndex));
     String databaseName = source.get("db");
     String tableName = source.get("table");
     SourceTable sourceTable = getSourceTable(databaseName, tableName);
@@ -200,7 +198,7 @@ private void handleDML(StructuredRecord source, StructuredRecord val, Map<String
 
     Long ingestTime = val.get("ts_ms");
     DMLEvent.Builder builder = DMLEvent.builder()
-      .setOffset(new Offset(deltaOffset))
+      .setOffset(recordOffset)
       .setOperationType(op)
       .setDatabaseName(databaseName)
       .setTableName(tableName)
@@ -218,9 +216,8 @@ private void handleDML(StructuredRecord source, StructuredRecord val, Map<String
     }
   }
 
-  private void handleDDL(String ddlStatement, Map<String, String> deltaOffset,
+  private void handleDDL(String ddlStatement, Offset recordOffset,
                          boolean isSnapshot, boolean readAllTables) throws InterruptedException {
-    deltaOffset.put(MySqlEventReader.SCHEMA_HISTORY_INDEX, String.valueOf(++schemaHistoryIndex));
     ddlParser.getDdlChanges().reset();
     ddlParser.parse(ddlStatement, tables);
     AtomicReference<InterruptedException> interrupted = new AtomicReference<>();
@@ -230,7 +227,7 @@ private void handleDDL(String ddlStatement, Map<String, String> deltaOffset,
       }
       for (DdlParserListener.Event event : events) {
         DDLEvent.Builder builder = DDLEvent.builder()
-          .setOffset(new Offset(deltaOffset))
+          .setOffset(recordOffset)
           .setDatabaseName(databaseName)
           .setSnapshot(isSnapshot);
         DDLEvent ddlEvent = null;

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) {