Merge pull request #75 from data-integrations/bugfix_release/CDAP-16735-fix-ddl-errors-during-snapshot

Bugfix release/cdap 16735 fix ddl errors during snapshot

Author: albertshau

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

@@ -40,6 +40,10 @@ public class DBSchemaHistory extends AbstractDatabaseHistory {
   private final DocumentWriter writer = DocumentWriter.defaultWriter();
   private final DocumentReader reader = DocumentReader.defaultReader();
 
+  public static void wipeHistory() throws IOException {
+    deltaRuntimeContext.putState(KEY, new byte[] { });
+  }
+
   @Override
   protected synchronized void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
     List<HistoryRecord> history = getHistory();
@@ -79,7 +83,7 @@ private List<HistoryRecord> getHistory() {
     List<HistoryRecord> history = new ArrayList<>();
     try {
       byte[] historyBytes = deltaRuntimeContext.getState(KEY);
-      if (historyBytes == null) {
+      if (historyBytes == null || historyBytes.length == 0) {
         return history;
       }
       String historyStr = Bytes.toString(historyBytes);

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.delta.plugin.mock;
+
+import io.cdap.delta.api.DDLEvent;
+import io.cdap.delta.api.DMLEvent;
+import io.cdap.delta.api.EventEmitter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * EventEmitter for tests that can block with emitting events.
+ */
+public class BlockingEventEmitter implements EventEmitter {
+  private final BlockingQueue<DDLEvent> ddlQueue;
+  private final BlockingQueue<DMLEvent> dmlQueue;
+
+  public BlockingEventEmitter(BlockingQueue<DDLEvent> ddlQueue, BlockingQueue<DMLEvent> dmlQueue) {
+    this.ddlQueue = ddlQueue;
+    this.dmlQueue = dmlQueue;
+  }
+
+  @Override
+  public void emit(DDLEvent ddlEvent) throws InterruptedException {
+    ddlQueue.put(ddlEvent);
+  }
+
+  @Override
+  public void emit(DMLEvent dmlEvent) throws InterruptedException {
+    dmlQueue.put(dmlEvent);
+  }
+}

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

@@ -16,6 +16,8 @@
 
 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;
@@ -37,6 +39,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.sql.Driver;
 import java.time.temporal.ChronoField;
 import java.time.temporal.ChronoUnit;
@@ -59,6 +62,7 @@ public class MySqlEventReader implements EventReader {
   private final DeltaSourceContext context;
   private final Set<SourceTable> sourceTables;
   private EmbeddedEngine engine;
+  private volatile boolean failedToStop;
 
   public MySqlEventReader(Set<SourceTable> sourceTables, MySqlConfig config,
                           DeltaSourceContext context, EventEmitter emitter) {
@@ -67,6 +71,7 @@ public MySqlEventReader(Set<SourceTable> sourceTables, MySqlConfig config,
     this.context = context;
     this.emitter = emitter;
     this.executorService = Executors.newSingleThreadScheduledExecutor();
+    this.failedToStop = false;
   }
 
   @Override
@@ -111,6 +116,20 @@ public void start(Offset offset) {
       .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
+     */
+    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);
     DdlParser ddlParser = mysqlConf.getDdlParsingMode().getNewParserInstance(mySqlValueConverters, tableId -> true);
@@ -132,10 +151,16 @@ public void start(Offset offset) {
   }
 
   public void stop() throws InterruptedException {
-    if (engine != null && engine.stop()) {
-      engine.await(1, TimeUnit.MINUTES);
+    executorService.shutdownNow();
+    if (!executorService.awaitTermination(2, TimeUnit.MINUTES)) {
+      LOG.warn("Unable to cleanly shutdown reader within the timeout.");
+      failedToStop = true;
     }
-    executorService.shutdown();
+  }
+
+  @VisibleForTesting
+  boolean failedToStop() {
+    return failedToStop;
   }
 
   private static MySqlValueConverters getValueConverters(MySqlConnectorConfig configuration) {

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

@@ -27,6 +27,7 @@
 import io.cdap.delta.api.SourceTable;
 import io.cdap.delta.plugin.common.Records;
 import io.debezium.connector.mysql.MySqlValueConverters;
+import io.debezium.embedded.StopConnectorException;
 import io.debezium.relational.Table;
 import io.debezium.relational.TableId;
 import io.debezium.relational.Tables;
@@ -40,6 +41,7 @@
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 /**
@@ -128,88 +130,28 @@ 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();
 
-    if (ddl != null) {
-      ddlParser.getDdlChanges().reset();
-      ddlParser.parse(ddl, tables);
-      ddlParser.getDdlChanges().groupEventsByDatabase((database, events) -> {
-        for (DdlParserListener.Event event : events) {
-          DDLEvent.Builder builder = DDLEvent.builder()
-            .setDatabase(database)
-            .setOffset(recordOffset)
-            .setSnapshot(isSnapshot);
-          // since current ddl blacklist implementation is bind with table level, we will only do the ddl blacklist
-          // checking only for table change related ddl event, includes: ALTER_TABLE, RENAME_TABLE, DROP_TABLE,
-          // CREATE_TABLE and TRUNCATE_TABLE.
-          switch (event.type()) {
-            case ALTER_TABLE:
-              DdlParserListener.TableAlteredEvent alteredEvent = (DdlParserListener.TableAlteredEvent) event;
-              TableId tableId = alteredEvent.tableId();
-              Table table = tables.forTable(tableId);
-              SourceTable sourceTable = getSourceTable(database, tableId.table());
-              DDLOperation ddlOp;
-              if (alteredEvent.previousTableId() != null) {
-                ddlOp = DDLOperation.RENAME_TABLE;
-                builder.setPrevTable(alteredEvent.previousTableId().table());
-              } else {
-                ddlOp = DDLOperation.ALTER_TABLE;
-              }
-
-              if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, ddlOp)) {
-                emitter.emit(builder.setOperation(ddlOp)
-                               .setTable(tableId.table())
-                               .setSchema(readAllTables ? Records.getSchema(table, mySqlValueConverters) :
-                                            Records.getSchema(table, mySqlValueConverters, sourceTable.getColumns()))
-                               .setPrimaryKey(table.primaryKeyColumnNames())
-                               .build());
-              }
-              break;
-            case DROP_TABLE:
-              DdlParserListener.TableDroppedEvent droppedEvent = (DdlParserListener.TableDroppedEvent) event;
-              sourceTable = getSourceTable(database, droppedEvent.tableId().table());
-              if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.DROP_TABLE)) {
-                emitter.emit(builder.setOperation(DDLOperation.DROP_TABLE)
-                               .setTable(droppedEvent.tableId().table())
-                               .build());
-              }
-              break;
-            case CREATE_TABLE:
-              DdlParserListener.TableCreatedEvent createdEvent = (DdlParserListener.TableCreatedEvent) event;
-              tableId = createdEvent.tableId();
-              table = tables.forTable(tableId);
-              sourceTable = getSourceTable(database, tableId.table());
-              if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.CREATE_TABLE)) {
-                emitter.emit(builder.setOperation(DDLOperation.CREATE_TABLE)
-                               .setTable(tableId.table())
-                               .setSchema(readAllTables ? Records.getSchema(table, mySqlValueConverters) :
-                                            Records.getSchema(table, mySqlValueConverters, sourceTable.getColumns()))
-                               .setPrimaryKey(table.primaryKeyColumnNames())
-                               .build());
-              }
-              break;
-            case DROP_DATABASE:
-              emitter.emit(builder.setOperation(DDLOperation.DROP_DATABASE).build());
-              break;
-            case CREATE_DATABASE:
-              emitter.emit(builder.setOperation(DDLOperation.CREATE_DATABASE).build());
-              break;
-            case TRUNCATE_TABLE:
-              DdlParserListener.TableTruncatedEvent truncatedEvent =
-                (DdlParserListener.TableTruncatedEvent) event;
-              sourceTable = getSourceTable(database, truncatedEvent.tableId().table());
-              if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.TRUNCATE_TABLE)) {
-                emitter.emit(builder.setOperation(DDLOperation.TRUNCATE_TABLE)
-                               .setTable(truncatedEvent.tableId().table())
-                               .build());
-              }
-              break;
-            default:
-              return;
-          }
-        }
-      });
-      return;
+    try {
+      if (ddl != null) {
+        handleDDL(ddl, recordOffset, isSnapshot, readAllTables);
+        return;
+      }
+
+      String databaseName = source.get("db");
+      String tableName = source.get("table");
+      SourceTable sourceTable = getSourceTable(databaseName, tableName);
+      if (sourceTableNotValid(readAllTables, sourceTable)) {
+        return;
+      }
+
+      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, Offset recordOffset,
+                         boolean isSnapshot, boolean readAllTables) throws InterruptedException {
     String databaseName = source.get("db");
     String tableName = source.get("table");
     SourceTable sourceTable = getSourceTable(databaseName, tableName);
@@ -274,6 +216,108 @@ public void accept(SourceRecord sourceRecord) {
     }
   }
 
+  private void handleDDL(String ddlStatement, Offset recordOffset,
+                         boolean isSnapshot, boolean readAllTables) throws InterruptedException {
+    ddlParser.getDdlChanges().reset();
+    ddlParser.parse(ddlStatement, tables);
+    AtomicReference<InterruptedException> interrupted = new AtomicReference<>();
+    ddlParser.getDdlChanges().groupEventsByDatabase((database, events) -> {
+      if (interrupted.get() != null) {
+        return;
+      }
+      for (DdlParserListener.Event event : events) {
+        DDLEvent.Builder builder = DDLEvent.builder()
+          .setOffset(recordOffset)
+          .setDatabase(database)
+          .setSnapshot(isSnapshot);
+        DDLEvent ddlEvent = null;
+        // since current ddl blacklist implementation is bind with table level, we will only do the ddl blacklist
+        // checking only for table change related ddl event, includes: ALTER_TABLE, RENAME_TABLE, DROP_TABLE,
+        // CREATE_TABLE and TRUNCATE_TABLE.
+        switch (event.type()) {
+          case ALTER_TABLE:
+            DdlParserListener.TableAlteredEvent alteredEvent = (DdlParserListener.TableAlteredEvent) event;
+            TableId tableId = alteredEvent.tableId();
+            Table table = tables.forTable(tableId);
+            SourceTable sourceTable = getSourceTable(database, tableId.table());
+            DDLOperation ddlOp;
+            if (alteredEvent.previousTableId() != null) {
+              ddlOp = DDLOperation.RENAME_TABLE;
+              builder.setPrevTable(alteredEvent.previousTableId().table());
+            } else {
+              ddlOp = DDLOperation.ALTER_TABLE;
+            }
+
+            if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, ddlOp)) {
+              ddlEvent = builder.setOperation(ddlOp)
+                .setTable(tableId.table())
+                .setSchema(readAllTables ? Records.getSchema(table, mySqlValueConverters) :
+                             Records.getSchema(table, mySqlValueConverters, sourceTable.getColumns()))
+                .setPrimaryKey(table.primaryKeyColumnNames())
+                .build();
+            }
+            break;
+          case DROP_TABLE:
+            DdlParserListener.TableDroppedEvent droppedEvent = (DdlParserListener.TableDroppedEvent) event;
+            sourceTable = getSourceTable(database, droppedEvent.tableId().table());
+            if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.DROP_TABLE)) {
+              ddlEvent = builder.setOperation(DDLOperation.DROP_TABLE)
+                .setTable(droppedEvent.tableId().table())
+                .build();
+            }
+            break;
+          case CREATE_TABLE:
+            DdlParserListener.TableCreatedEvent createdEvent = (DdlParserListener.TableCreatedEvent) event;
+            tableId = createdEvent.tableId();
+            table = tables.forTable(tableId);
+            sourceTable = getSourceTable(database, tableId.table());
+            if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.CREATE_TABLE)) {
+              ddlEvent = builder.setOperation(DDLOperation.CREATE_TABLE)
+                .setTable(tableId.table())
+                .setSchema(readAllTables ? Records.getSchema(table, mySqlValueConverters) :
+                             Records.getSchema(table, mySqlValueConverters, sourceTable.getColumns()))
+                .setPrimaryKey(table.primaryKeyColumnNames())
+                .build();
+            }
+            break;
+          case DROP_DATABASE:
+            ddlEvent = builder.setOperation(DDLOperation.DROP_DATABASE).build();
+            break;
+          case CREATE_DATABASE:
+            // due to a bug in io.debezium.relational.ddl.AbstractDdlParser#signalDropDatabase
+            // a DROP_DATABASE event will be mistakenly categorized as a CREATE_DATABASE event.
+            // TODO: check if this is fixed in a newer debezium version
+            if (event.statement() != null && event.statement().startsWith("DROP DATABASE")) {
+              ddlEvent = builder.setOperation(DDLOperation.DROP_DATABASE).build();
+            } else {
+              ddlEvent = builder.setOperation(DDLOperation.CREATE_DATABASE).build();
+            }
+            break;
+          case TRUNCATE_TABLE:
+            DdlParserListener.TableTruncatedEvent truncatedEvent =
+              (DdlParserListener.TableTruncatedEvent) event;
+            sourceTable = getSourceTable(database, truncatedEvent.tableId().table());
+            if (shouldEmitDdlEventForOperation(readAllTables, sourceTable, DDLOperation.TRUNCATE_TABLE)) {
+              ddlEvent = builder.setOperation(DDLOperation.TRUNCATE_TABLE)
+                .setTable(truncatedEvent.tableId().table())
+                .build();
+            }
+            break;
+        }
+        if (ddlEvent != null) {
+          try {
+            emitter.emit(ddlEvent);
+          } catch (InterruptedException e) {
+            interrupted.set(e);
+          }
+        }
+      }
+    });
+    if (interrupted.get() != null) {
+      throw interrupted.get();
+    }
+  }
+
   private boolean shouldEmitDdlEventForOperation(boolean readAllTables, SourceTable sourceTable, DDLOperation op) {
     return (!sourceTableNotValid(readAllTables, sourceTable)) &&
       (!isDDLOperationBlacklisted(readAllTables, sourceTable, op));