Merge branch 'develop' of github.com:data-integrations/database-delta-plugins into develop

Author: sagarkapare

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

@@ -37,9 +37,21 @@ public class DBSchemaHistory extends AbstractDatabaseHistory {
   private static final String KEY = "history";
   // Hacky, fix when usage of EmbeddedEngine is replaced
   public static DeltaRuntimeContext deltaRuntimeContext;
-  private final DocumentWriter writer = DocumentWriter.defaultWriter();
-  private final DocumentReader reader = DocumentReader.defaultReader();
+  private static final DocumentWriter writer = DocumentWriter.defaultWriter();
+  private static 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[] { });
   }
@@ -48,6 +60,10 @@ public static void wipeHistory() throws IOException {
   protected synchronized void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
     List<HistoryRecord> history = getHistory();
     history.add(record);
+    storeHistory(history);
+  }
+
+  private static void storeHistory(List<HistoryRecord> history) {
     String historyStr = history.stream().map(r -> {
       try {
         return writer.write(r.document());
@@ -84,7 +100,7 @@ public boolean storageExists() {
   }
 
   // TODO: cache history, should only have to read once
-  private List<HistoryRecord> getHistory() {
+  private static 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

@@ -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;
@@ -39,6 +38,7 @@
 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,6 +59,8 @@
  */
 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;
@@ -124,19 +126,16 @@ public void start(Offset offset) {
     Configuration debeziumConf = configBuilder.build();
     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
+    /**
+     * 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.
      */
-    if (offset.get().isEmpty()) {
-      try {
-        DBSchemaHistory.wipeHistory();
-      } catch (IOException e) {
-        throw new RuntimeException("Unable to wipe schema history at start of replication.", e);
-      }
+    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);
     }
 
     MySqlValueConverters mySqlValueConverters = getValueConverters(mysqlConf);
@@ -149,7 +148,7 @@ public void start(Offset offset) {
       engine = EmbeddedEngine.create()
         .using(debeziumConf)
         .notifying(new MySqlRecordConsumer(context, emitter, ddlParser, mySqlValueConverters,
-                                           new Tables(), sourceTableMap))
+                                           new Tables(), sourceTableMap, schemaHistoryIndex))
         .using(new NotifyingCompletionCallback(context))
         .build();
       executorService.submit(engine);

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

@@ -56,16 +56,18 @@ 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) {
+                             Tables tables, Map<String, SourceTable> sourceTableMap, int schemaHistoryIndex) {
     this.context = context;
     this.emitter = emitter;
     this.ddlParser = ddlParser;
     this.mySqlValueConverters = mySqlValueConverters;
     this.tables = tables;
     this.sourceTableMap = sourceTableMap;
+    this.schemaHistoryIndex = schemaHistoryIndex;
   }
 
   @Override
@@ -116,11 +118,8 @@ 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
@@ -130,9 +129,11 @@ public void accept(SourceRecord sourceRecord) {
     // 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, recordOffset, isSnapshot, readAllTables);
+        handleDDL(ddl, deltaOffset, isSnapshot, readAllTables);
         return;
       }
 
@@ -143,15 +144,16 @@ public void accept(SourceRecord sourceRecord) {
         return;
       }
 
-      handleDML(source, val, recordOffset, isSnapshot, readAllTables);
+      handleDML(source, val, deltaOffset, 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, Offset recordOffset,
+  private void handleDML(StructuredRecord source, StructuredRecord val, Map<String, String> deltaOffset,
                          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);
@@ -198,7 +200,7 @@ private void handleDML(StructuredRecord source, StructuredRecord val, Offset rec
 
     Long ingestTime = val.get("ts_ms");
     DMLEvent.Builder builder = DMLEvent.builder()
-      .setOffset(recordOffset)
+      .setOffset(new Offset(deltaOffset))
       .setOperationType(op)
       .setDatabaseName(databaseName)
       .setTableName(tableName)
@@ -216,8 +218,9 @@ private void handleDML(StructuredRecord source, StructuredRecord val, Offset rec
     }
   }
 
-  private void handleDDL(String ddlStatement, Offset recordOffset,
+  private void handleDDL(String ddlStatement, Map<String, String> deltaOffset,
                          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<>();
@@ -227,7 +230,7 @@ private void handleDDL(String ddlStatement, Offset recordOffset,
       }
       for (DdlParserListener.Event event : events) {
         DDLEvent.Builder builder = DDLEvent.builder()
-          .setOffset(recordOffset)
+          .setOffset(new Offset(deltaOffset))
           .setDatabaseName(databaseName)
           .setSnapshot(isSnapshot);
         DDLEvent ddlEvent = null;