CDAP-17584 clear the snapshotting table list when resume from middle of the snapshotting (#140)

Author: seanzhougoogle

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

@@ -125,16 +125,17 @@ public void start(Offset offset) {
 
     DBSchemaHistory.deltaRuntimeContext = context;
 
-    String snapshotTablesStr = state.get(SqlServerOffset.SNAPSHOT_TABLES);
-    Set<String> snapshotTables = Strings.isNullOrEmpty(snapshotTablesStr) ? new HashSet<>() :
-      new HashSet<>(Arrays.asList(snapshotTablesStr.split(SqlServerOffset.DELIMITER)));
+    String ddlEventSentStr = state.get(SqlServerOffset.DDL_EVENT_SENT);
+    Set<String> ddlEventSent = Strings.isNullOrEmpty(ddlEventSentStr) ? new HashSet<>() :
+      new HashSet<>(Arrays.asList(ddlEventSentStr.split(SqlServerOffset.DELIMITER)));
 
     /*
      * 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() || !"true".equalsIgnoreCase(isSnapshotCompleted)) {
+      ddlEventSent.clear();
       try {
         DBSchemaHistory.wipeHistory();
       } catch (IOException e) {
@@ -149,7 +150,7 @@ public void start(Offset offset) {
       LOG.info("creating new EmbeddedEngine...");
       // Create the engine with this configuration ...
       engine = EmbeddedEngine.create()
-        .notifying(new SqlServerRecordConsumer(context, emitter, databaseName, snapshotTables, sourceTableMap, offset))
+        .notifying(new SqlServerRecordConsumer(context, emitter, databaseName, ddlEventSent, sourceTableMap, offset))
         .using(debeziumConf)
         .using(new NotifyingCompletionCallback(context))
         .build();

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

@@ -31,32 +31,32 @@
  */
 public class SqlServerOffset {
   static final String DELIMITER = ",";
-  static final String SNAPSHOT_TABLES = "snapshot_tables";
+  static final String DDL_EVENT_SENT = "ddl_event_sent";
 
   private final String changeLsn;
   private final String commitLsn;
   private final Boolean isSnapshot;
   private final Boolean isSnapshotCompleted;
-  private Set<String> snapshotTables;
+  private Set<String> ddlEventSent;
 
   SqlServerOffset(Map<String, ?> properties) {
     this.changeLsn = (String) properties.get(SourceInfo.CHANGE_LSN_KEY);
     this.commitLsn = (String) properties.get(SourceInfo.COMMIT_LSN_KEY);
     this.isSnapshot = (Boolean) properties.get(SourceInfo.SNAPSHOT_KEY);
     this.isSnapshotCompleted = (Boolean) properties.get(SqlServerConstantOffsetBackingStore.SNAPSHOT_COMPLETED);
-    this.snapshotTables = new HashSet<>();
+    this.ddlEventSent = new HashSet<>();
   }
 
   boolean isSnapshot() {
     return isSnapshot;
   }
 
-  void setSnapshotTables(Set<String> snapshotTables) {
-    this.snapshotTables = new HashSet<>(snapshotTables);
+  void setDdlEventSent(Set<String> ddlEventSent) {
+    this.ddlEventSent = new HashSet<>(ddlEventSent);
   }
 
   void addSnapshotTable(String table) {
-    snapshotTables.add(table);
+    ddlEventSent.add(table);
   }
 
   Offset getAsOffset() {
@@ -73,8 +73,8 @@ Offset getAsOffset() {
     if (isSnapshotCompleted != null) {
       deltaOffset.put(SqlServerConstantOffsetBackingStore.SNAPSHOT_COMPLETED, String.valueOf(isSnapshotCompleted));
     }
-    if (snapshotTables != null && !snapshotTables.isEmpty()) {
-      deltaOffset.put(SNAPSHOT_TABLES, String.join(DELIMITER, snapshotTables));
+    if (ddlEventSent != null && !ddlEventSent.isEmpty()) {
+      deltaOffset.put(DDL_EVENT_SENT, String.join(DELIMITER, ddlEventSent));
     }
 
     return new Offset(deltaOffset);
@@ -106,11 +106,11 @@ public boolean equals(Object o) {
       && Objects.equals(commitLsn, that.commitLsn)
       && Objects.equals(isSnapshot, that.isSnapshot)
       && Objects.equals(isSnapshotCompleted, that.isSnapshotCompleted)
-      && Objects.equals(snapshotTables, that.snapshotTables);
+      && Objects.equals(ddlEventSent, that.ddlEventSent);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(changeLsn, commitLsn, isSnapshot, isSnapshotCompleted, snapshotTables);
+    return Objects.hash(changeLsn, commitLsn, isSnapshot, isSnapshotCompleted, ddlEventSent);
   }
 }

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

@@ -51,18 +51,19 @@ public class SqlServerRecordConsumer implements Consumer<SourceRecord> {
   private final EventEmitter emitter;
   // we need this since there is no way to get the db information from the source record
   private final String databaseName;
-  private final Set<String> snapshotTables;
+  // record tables that already had DDL events sent
+  private final Set<String> ddlEventSent;
   private final Map<String, SourceTable> sourceTableMap;
   private Offset latestOffset;
 
 
   SqlServerRecordConsumer(DeltaSourceContext context, EventEmitter emitter, String databaseName,
-                          Set<String> snapshotTables, Map<String, SourceTable> sourceTableMap,
+                          Set<String> ddlEventSent, Map<String, SourceTable> sourceTableMap,
     Offset latestOffset) {
     this.context = context;
     this.emitter = emitter;
     this.databaseName = databaseName;
-    this.snapshotTables = snapshotTables;
+    this.ddlEventSent = ddlEventSent;
     this.sourceTableMap = sourceTableMap;
     this.latestOffset = latestOffset;
   }
@@ -89,7 +90,7 @@ public void accept(SourceRecord sourceRecord) {
       return;
     }
 
-    sqlServerOffset.setSnapshotTables(snapshotTables);
+    sqlServerOffset.setDdlEventSent(ddlEventSent);
     boolean isSnapshot = sqlServerOffset.isSnapshot();
     latestOffset = sqlServerOffset.getAsOffset();
 
@@ -149,7 +150,7 @@ public void accept(SourceRecord sourceRecord) {
     Schema schema = value.getSchema();
     // send the ddl events only if we see the table at the first time
     // Note: the delta app itself have prevented adding CREATE_TABLE operation into DDL blacklist for all the tables.
-    if (!snapshotTables.contains(sourceTableId)) {
+    if (!ddlEventSent.contains(sourceTableId)) {
       StructuredRecord key = Records.convert((Struct) sourceRecord.key());
       List<Schema.Field> fields = key.getSchema().getFields();
       List<String> primaryFields = new ArrayList<>();
@@ -182,7 +183,7 @@ public void accept(SourceRecord sourceRecord) {
         // happens when the event reader is stopped. throwing this exception tells Debezium to stop right away
         throw new StopConnectorException("Interrupted while emitting an event.");
       }
-      snapshotTables.add(sourceTableId);
+      ddlEventSent.add(sourceTableId);
     }
 
     if (!readAllTables && sourceTable.getDmlBlacklist().contains(op)) {