CDAP-16737 fix stop logic

Fixed stop logic for the event readers. Event consumers need to
throw a special StopConnectorException to tell Debezium to stop
immediately instead of continuing. Fixed the consumers to do this
if they are interrupted while writing to the event queue, which
happens when the reader is being stopped.

Also added integration tests for stopping logic and
included fixes to mysql integration test

Author: albertshau

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

@@ -16,6 +16,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;
@@ -61,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) {
@@ -69,6 +71,7 @@ public MySqlEventReader(Set<SourceTable> sourceTables, MySqlConfig config,
     this.context = context;
     this.emitter = emitter;
     this.executorService = Executors.newSingleThreadScheduledExecutor();
+    this.failedToStop = false;
   }
 
   @Override
@@ -151,9 +154,15 @@ public void stop() throws InterruptedException {
     executorService.shutdownNow();
     if (!executorService.awaitTermination(2, TimeUnit.MINUTES)) {
       LOG.warn("Unable to cleanly shutdown reader within the timeout.");
+      failedToStop = true;
     }
   }
 
+  @VisibleForTesting
+  boolean failedToStop() {
+    return failedToStop;
+  }
+
   private static MySqlValueConverters getValueConverters(MySqlConnectorConfig configuration) {
     // Use MySQL-specific converters and schemas for values ...
 

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;
@@ -144,9 +145,8 @@ public void accept(SourceRecord sourceRecord) {
 
       handleDML(source, val, recordOffset, isSnapshot, readAllTables);
     } catch (InterruptedException e) {
-      LOG.debug("Interrupted while emitting change event.", e);
-      // reset the interrupted flag so that caller knows to interrupt
-      Thread.currentThread().interrupt();
+      // happens when the event reader is stopped. throwing this exception tells Debezium to stop right away
+      throw new StopConnectorException("Interrupted while emitting event.");
     }
   }
 
@@ -284,7 +284,14 @@ private void handleDDL(String ddlStatement, Offset recordOffset,
             ddlEvent = builder.setOperation(DDLOperation.DROP_DATABASE).build();
             break;
           case CREATE_DATABASE:
-            ddlEvent = builder.setOperation(DDLOperation.CREATE_DATABASE).build();
+            // 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 =

mysql-delta-plugins/src/test/java/io/cdap/delta/mysql/MySqlEventReaderIntegrationTest.java

@@ -24,8 +24,10 @@
 import io.cdap.delta.api.DMLEvent;
 import io.cdap.delta.api.DMLOperation;
 import io.cdap.delta.api.DeltaSourceContext;
+import io.cdap.delta.api.EventEmitter;
 import io.cdap.delta.api.Offset;
 import io.cdap.delta.api.SourceTable;
+import io.cdap.delta.plugin.mock.BlockingEventEmitter;
 import io.cdap.delta.plugin.mock.MockContext;
 import io.cdap.delta.plugin.mock.MockEventEmitter;
 import org.junit.Assert;
@@ -44,6 +46,8 @@
 import java.util.Collections;
 import java.util.Properties;
 import java.util.TimeZone;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -100,8 +104,9 @@ public static void setupClass() throws Exception {
     try (Connection connection = DriverManager.getConnection(connectionUrl, connProperties)) {
       // create table
       try (Statement statement = connection.createStatement()) {
-        statement.execute(String.format("CREATE TABLE %s (id int PRIMARY KEY, name varchar(50), bday date null)",
-                                        CUSTOMERS_TABLE));
+        statement.execute(
+          String.format("CREATE TABLE %s (id int PRIMARY KEY, name varchar(50) not null, bday date null)",
+                        CUSTOMERS_TABLE));
       }
 
       // insert sample data
@@ -142,30 +147,25 @@ public void test() throws InterruptedException {
     Assert.assertEquals(4, eventEmitter.getDdlEvents().size());
     Assert.assertEquals(2, eventEmitter.getDmlEvents().size());
 
-    /*
-      This currently would fails.
-      Need to investigate why it's DROP_TABLE, CREATE_DATABASE, CREATE_DATABASE, CREATE_TABLE
     DDLEvent ddlEvent = eventEmitter.getDdlEvents().get(0);
-    Assert.assertEquals(DDLOperation.DROP_DATABASE, ddlEvent.getOperation());
+    Assert.assertEquals(DDLOperation.DROP_TABLE, ddlEvent.getOperation());
     Assert.assertEquals(DB, ddlEvent.getDatabase());
+    Assert.assertEquals(CUSTOMERS_TABLE, ddlEvent.getTable());
 
     ddlEvent = eventEmitter.getDdlEvents().get(1);
-    Assert.assertEquals(DDLOperation.CREATE_DATABASE, ddlEvent.getOperation());
+    Assert.assertEquals(DDLOperation.DROP_DATABASE, ddlEvent.getOperation());
     Assert.assertEquals(DB, ddlEvent.getDatabase());
 
     ddlEvent = eventEmitter.getDdlEvents().get(2);
-    Assert.assertEquals(DDLOperation.DROP_TABLE, ddlEvent.getOperation());
+    Assert.assertEquals(DDLOperation.CREATE_DATABASE, ddlEvent.getOperation());
     Assert.assertEquals(DB, ddlEvent.getDatabase());
-    Assert.assertEquals(CUSTOMERS_TABLE, ddlEvent.getTable());
-    */
 
-    DDLEvent ddlEvent = eventEmitter.getDdlEvents().get(3);
+    ddlEvent = eventEmitter.getDdlEvents().get(3);
     Assert.assertEquals(DDLOperation.CREATE_TABLE, ddlEvent.getOperation());
     Assert.assertEquals(DB, ddlEvent.getDatabase());
     Assert.assertEquals(CUSTOMERS_TABLE, ddlEvent.getTable());
     Assert.assertEquals(Collections.singletonList("id"), ddlEvent.getPrimaryKey());
-    // TODO: this would currently fail, seems to incorrectly return the schema as non-nullable
-    //Assert.assertEquals(CUSTOMERS_SCHEMA, ddlEvent.getSchema());
+    Assert.assertEquals(CUSTOMERS_SCHEMA, ddlEvent.getSchema());
 
     DMLEvent dmlEvent = eventEmitter.getDmlEvents().get(0);
     Assert.assertEquals(DMLOperation.INSERT, dmlEvent.getOperation());
@@ -177,8 +177,7 @@ public void test() throws InterruptedException {
       .set("name", "alice")
       .setDate("bday", LocalDate.ofEpochDay(0))
       .build();
-    // this fails with schemas that are different
-    // Assert.assertEquals(expected, row);
+    Assert.assertEquals(expected, row);
 
     dmlEvent = eventEmitter.getDmlEvents().get(1);
     Assert.assertEquals(DMLOperation.INSERT, dmlEvent.getOperation());
@@ -190,7 +189,37 @@ public void test() throws InterruptedException {
       .set("name", "bob")
       .setDate("bday", LocalDate.ofEpochDay(365))
       .build();
-    // this fails with schemas that are different
-    // Assert.assertEquals(expected, row);
+    Assert.assertEquals(expected, row);
+  }
+
+  @Test
+  public void stopReaderTest() throws Exception {
+    SourceTable sourceTable = new SourceTable(DB, CUSTOMERS_TABLE, null,
+                                              Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
+
+    DeltaSourceContext context = new MockContext(Driver.class);
+    BlockingQueue<DDLEvent> ddlEvents = new ArrayBlockingQueue<>(1);
+    BlockingQueue<DMLEvent> dmlEvents = new ArrayBlockingQueue<>(1);
+    EventEmitter eventEmitter = new BlockingEventEmitter(ddlEvents, dmlEvents);
+    MySqlConfig config = new MySqlConfig("localhost", port, "root", password, 13, DB,
+                                         TimeZone.getDefault().getID());
+
+    MySqlEventReader eventReader = new MySqlEventReader(Collections.singleton(sourceTable), config,
+                                                        context, eventEmitter);
+
+    eventReader.start(new Offset());
+
+    int count = 0;
+    while (ddlEvents.size() < 1 && count < 100) {
+      TimeUnit.MILLISECONDS.sleep(50);
+      count++;
+    }
+
+    if (count >= 100) {
+      Assert.fail("Reader never emitted any events.");
+    }
+
+    eventReader.stop();
+    Assert.assertFalse(eventReader.failedToStop());
   }
 }

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

@@ -59,7 +59,6 @@ public void configure(Configurer configurer) {
 
   @Override
   public EventReader createReader(EventReaderDefinition definition, DeltaSourceContext context, EventEmitter emitter) {
-    LOG.info("creating new event reader");
     return new SqlServerEventReader(definition.getTables(), config, context, emitter);
   }
 

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

@@ -28,6 +28,7 @@
 import io.cdap.delta.api.SourceTable;
 import io.cdap.delta.plugin.common.Records;
 import io.debezium.connector.sqlserver.SourceInfo;
+import io.debezium.embedded.StopConnectorException;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.slf4j.Logger;
@@ -63,7 +64,6 @@ public class SqlServerRecordConsumer implements Consumer<SourceRecord> {
     this.databaseName = databaseName;
     this.trackingTables = new HashSet<>();
     this.sourceTableMap = sourceTableMap;
-    LOG.info("created new record consumer.");
   }
 
   @Override
@@ -80,14 +80,8 @@ public void accept(SourceRecord sourceRecord) {
     Map<String, String> deltaOffset = SqlServerConstantOffsetBackingStore.serializeOffsets(sourceRecord);
     Offset recordOffset = new Offset(deltaOffset);
     StructuredRecord val = Records.convert((Struct) sourceRecord.value());
-    Object snapshotVal = sourceRecord.sourceOffset().get(SourceInfo.SNAPSHOT_KEY);
-    LOG.info("snapshot val = {}", snapshotVal);
 
     boolean isSnapshot = Boolean.TRUE.equals(sourceRecord.sourceOffset().get(SourceInfo.SNAPSHOT_KEY));
-    LOG.info("{}, offset:", this);
-    for (Map.Entry<String, String> entry : deltaOffset.entrySet()) {
-      LOG.info(" {} = {}", entry.getKey(), entry.getValue());
-    }
 
     DMLOperation op;
     String opStr = val.get("op");
@@ -146,8 +140,6 @@ public void accept(SourceRecord sourceRecord) {
                                  .setSnapshot(isSnapshot);
 
     Schema schema = value.getSchema();
-    LOG.info("isSnapshot = {}", isSnapshot);
-    LOG.info("tracking tables has {} elements, = {}", trackingTables.size(), trackingTables);
     // send the ddl event if the first see the table and the it is in snapshot.
     // Note: the delta app itself have prevented adding CREATE_TABLE operation into DDL blacklist for all the tables.
     if (!trackingTables.contains(trackingTable) && isSnapshot) {
@@ -172,9 +164,8 @@ public void accept(SourceRecord sourceRecord) {
                        .setPrimaryKey(primaryFields)
                        .build());
       } catch (InterruptedException e) {
-        LOG.debug("Interrupted while emitting change event.", e);
-        Thread.currentThread().interrupt();
-        return;
+        // happens when the event reader is stopped. throwing this exception tells Debezium to stop right away
+        throw new StopConnectorException("Interrupted while emitting an event.");
       }
       trackingTables.add(trackingTable);
     }
@@ -203,8 +194,8 @@ public void accept(SourceRecord sourceRecord) {
     try {
       emitter.emit(dmlBuilder.build());
     } catch (InterruptedException e) {
-      LOG.debug("Interrupted while emitting change event.", e);
-      Thread.currentThread().interrupt();
+      // happens when the event reader is stopped. throwing this exception tells Debezium to stop right away
+      throw new StopConnectorException("Interrupted while emitting an event.");
     }
   }
 }

sqlserver-delta-plugins/src/test/java/io.cdap.delta.sqlserver/SqlServerEventReaderIntegrationTest.java

@@ -30,7 +30,6 @@
 import io.cdap.delta.plugin.mock.BlockingEventEmitter;
 import io.cdap.delta.plugin.mock.MockContext;
 import io.cdap.delta.plugin.mock.MockEventEmitter;
-import org.eclipse.jetty.util.BlockingArrayQueue;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -46,6 +45,7 @@
 import java.time.LocalDate;
 import java.util.Collections;
 import java.util.Properties;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 
@@ -202,8 +202,8 @@ public void testEventReaderStop() throws Exception {
                                               Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
 
     DeltaSourceContext context = new MockContext(SQLServerDriver.class);
-    BlockingQueue<DDLEvent> ddlEvents = new BlockingArrayQueue<>(1);
-    BlockingQueue<DMLEvent> dmlEvents = new BlockingArrayQueue<>(1);
+    BlockingQueue<DDLEvent> ddlEvents = new ArrayBlockingQueue<>(1);
+    BlockingQueue<DMLEvent> dmlEvents = new ArrayBlockingQueue<>(1);
     EventEmitter eventEmitter = new BlockingEventEmitter(ddlEvents, dmlEvents);
     SqlServerConfig config = new SqlServerConfig("localhost", port, "sa", password,
                                                  DB, null, "mssql");
@@ -213,9 +213,16 @@ public void testEventReaderStop() throws Exception {
 
     eventReader.start(new Offset());
 
-    while (ddlEvents.size() < 1) {
-      TimeUnit.MILLISECONDS.sleep(10);
+    int count = 0;
+    while (ddlEvents.size() < 1 && count < 100) {
+      TimeUnit.MILLISECONDS.sleep(50);
+      count++;
     }
+
+    if (count >= 100) {
+      Assert.fail("Reader never emitted any events.");
+    }
+
     eventReader.stop();
     Assert.assertFalse(eventReader.failedToStop());
   }