Revert "CDAP-16735 fix DBSchemaHistory"

This reverts commit 5b4458f.

Author: seanzhougoogle

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

@@ -37,21 +37,9 @@ 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[] { });
   }
@@ -60,10 +48,6 @@ 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());
@@ -100,7 +84,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

@@ -17,6 +17,7 @@
 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;
@@ -38,7 +39,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 +59,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;
@@ -126,16 +124,19 @@ 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.
+    /*
+       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
      */
-    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()) {
+      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 +149,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,8 +116,11 @@ 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
@@ -129,11 +130,9 @@ 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, 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;