Add delete test (#31)

Add delete test
memiiso · Aug 16, 2021 · 8d9dbb0 · 8d9dbb0
1 parent c87f058
commit 8d9dbb0
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 20 deletions.
diff --git a/...batch-sinks/src/test/java/io/debezium/server/batch/BatchSparkChangeConsumerMysqlTest.java b/...batch-sinks/src/test/java/io/debezium/server/batch/BatchSparkChangeConsumerMysqlTest.java
@@ -39,24 +39,35 @@ public class BatchSparkChangeConsumerMysqlTest extends BaseSparkTest {
   @ConfigProperty(name = "debezium.source.max.batch.size", defaultValue = "1000")
   Integer maxBatchSize;
 
-//  @Test
-//  public void testPerformance() throws Exception {
-//
-//    int iteration = 10;
-//    mysqlCreateTestDataTable();
-//    for (int i = 0; i <= iteration; i++) {
-//      mysqlLoadTestDataTable(maxBatchSize);
-//    }
-//
-//    Awaitility.await().atMost(Duration.ofSeconds(120)).until(() -> {
-//      try {
-//        Dataset<Row> df = getTableData("testc.inventory.test_date_table");
-//        return df.count() >= (long) iteration * maxBatchSize;
-//      } catch (Exception e) {
-//        return false;
-//      }
-//    });
-//  }
+  @Test
+  public void testTombstoneEvents() throws Exception {
+    // create test table
+    String sqlCreate = "CREATE TABLE IF NOT EXISTS inventory.test_delete_table (" +
+        " c_id INTEGER ," +
+        " c_id2 INTEGER ," +
+        " c_data TEXT," +
+        "  PRIMARY KEY (c_id, c_id2)" +
+        " );";
+    String sqlInsert =
+        "INSERT INTO inventory.test_delete_table (c_id, c_id2, c_data ) " +
+            "VALUES  (1,1,'data'),(1,2,'data'),(1,3,'data'),(1,4,'data') ;";
+    String sqlDelete = "DELETE FROM inventory.test_delete_table where c_id = 1 ;";
+
+    SourceMysqlDB.runSQL(sqlCreate);
+    SourceMysqlDB.runSQL(sqlInsert);
+    SourceMysqlDB.runSQL(sqlDelete);
+    SourceMysqlDB.runSQL(sqlInsert);
+
+    Awaitility.await().atMost(Duration.ofSeconds(120)).until(() -> {
+      try {
+        Dataset<Row> df = getTableData("testc.inventory.test_delete_table");
+        df.show();
+        return df.count() >= 12; // 4 insert 4 delete 4 insert!
+      } catch (Exception e) {
+        return false;
+      }
+    });
+  }
 
   @Test
   public void testSimpleUpload() {

diff --git a/debezium-server-batch-sinks/src/test/java/io/debezium/server/batch/ConfigSource.java b/debezium-server-batch-sinks/src/test/java/io/debezium/server/batch/ConfigSource.java
@@ -69,8 +69,8 @@ public ConfigSource() {
     config.put("debezium.transforms", "unwrap");
     config.put("debezium.transforms.unwrap.type", "io.debezium.transforms.ExtractNewRecordState");
     config.put("debezium.transforms.unwrap.add.fields", "op,table,source.ts_ms,db");
-    //s3Test.put("debezium.transforms.unwrap.add.headers", "db");
     config.put("debezium.transforms.unwrap.delete.handling.mode", "rewrite");
+    config.put("debezium.transforms.unwrap.drop.tombstones", "true");
 
     // DEBEZIUM SOURCE conf
     config.put("debezium.source.connector.class", "io.debezium.connector.postgresql.PostgresConnector");
@@ -86,7 +86,8 @@ public ConfigSource() {
     config.put("%mysql.debezium.source.database.include.list", "inventory");
     config.put("%postgresql.debezium.source.schema.include.list", "inventory");
     config.put("debezium.source.table.include.list", "inventory.customers,inventory.orders,inventory.products," +
-        "inventory.test_date_table,inventory.products_on_hand,inventory.geom,inventory.table_datatypes");
+        "inventory.products_on_hand,inventory.geom," +
+        "inventory.test_date_table,inventory.table_datatypes,inventory.test_delete_table");
 
     config.put("debezium.source.snapshot.select.statement.overrides.inventory.products_on_hand", "SELECT * FROM products_on_hand WHERE 1>2");
 //    However, when decimal.handling.mode configuration property is set to double, then the connector will represent