同步测试类

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/Common.java

@@ -1,13 +1,18 @@
 package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 import com.alibaba.otter.canal.client.adapter.es.test.TestConstant;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
 
-import java.util.HashMap;
-import java.util.Map;
-
 public class Common {
 
     public static ESAdapter init() {
@@ -24,4 +29,40 @@ public class Common {
         esAdapter.init(outerAdapterConfig);
         return esAdapter;
     }
+
+    public static void sqlExe(DataSource dataSource, String sql) {
+        Connection conn = null;
+        Statement stmt = null;
+        try {
+            conn = dataSource.getConnection();
+            conn.setAutoCommit(false);
+            stmt = conn.createStatement();
+            stmt.execute(sql);
+            conn.commit();
+        } catch (Exception e) {
+            if (conn != null) {
+                try {
+                    conn.rollback();
+                } catch (SQLException e1) {
+                    // ignore
+                }
+            }
+            e.printStackTrace();
+        } finally {
+            if (stmt != null) {
+                try {
+                    stmt.close();
+                } catch (SQLException e) {
+                    // ignore
+                }
+            }
+            if (conn != null) {
+                try {
+                    conn.close();
+                } catch (SQLException e) {
+                    // ignore
+                }
+            }
+        }
+    }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java

@@ -14,6 +14,8 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
+import javax.sql.DataSource;
+
 public class LabelSyncJoinSub2Test {
 
     private ESAdapter esAdapter;
@@ -25,10 +27,15 @@ public class LabelSyncJoinSub2Test {
     }
 
     /**
-     * 子查询从表插入 (确保主表记录必须有数据)
+     * 带函数子查询从表插入
      */
     @Test
-    public void insertTest01() {
+    public void test01() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"delete from label where id=1 or id=2");
+        Common.sqlExe(ds,"insert into label (id,user_id,label) values (1,1,'a')");
+        Common.sqlExe(ds,"insert into label (id,user_id,label) values (2,1,'b')");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -38,20 +45,26 @@ public class LabelSyncJoinSub2Test {
         List<Map<String, Object>> dataList = new ArrayList<>();
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
-        data.put("id", 1L);
+        data.put("id", 2L);
         data.put("user_id",1L);
-        data.put("label", "a");
+        data.put("label", "b");
 
         dml.setData(dataList);
 
         esAdapter.getEsSyncService().sync(dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
-        Assert.assertEquals("a;b_", response.getSource().get("_labels"));
+        Assert.assertEquals("b;a_", response.getSource().get("_labels"));
     }
 
+    /**
+     * 带函数子查询从表更新
+     */
     @Test
-    public void updateTest02() {
+    public void test02() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"update label set label='aa' where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -75,11 +88,17 @@ public class LabelSyncJoinSub2Test {
         esAdapter.getEsSyncService().sync(dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
-        Assert.assertEquals("aa;b_", response.getSource().get("_labels"));
+        Assert.assertEquals("b;aa_", response.getSource().get("_labels"));
     }
 
+    /**
+     * 带函数子查询从表删除
+     */
     @Test
-    public void deleteTest03() {
+    public void test03() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"delete from label where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -100,10 +119,4 @@ public class LabelSyncJoinSub2Test {
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b_", response.getSource().get("_labels"));
     }
-
-    @After
-    public void after() {
-        esAdapter.destroy();
-        DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
-    }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java

@@ -14,6 +14,8 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
+import javax.sql.DataSource;
+
 public class LabelSyncJoinSubTest {
 
     private ESAdapter esAdapter;
@@ -25,10 +27,15 @@ public class LabelSyncJoinSubTest {
     }
 
     /**
-     * 子查询从表插入 (确保主表记录必须有数据)
+     * 子查询从表插入
      */
     @Test
-    public void insertTest01() {
+    public void test01() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"delete from label where id=1 or id=2");
+        Common.sqlExe(ds,"insert into label (id,user_id,label) values (1,1,'a')");
+        Common.sqlExe(ds,"insert into label (id,user_id,label) values (2,1,'b')");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -38,9 +45,9 @@ public class LabelSyncJoinSubTest {
         List<Map<String, Object>> dataList = new ArrayList<>();
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
-        data.put("id", 1L);
+        data.put("id", 2L);
         data.put("user_id",1L);
-        data.put("label", "a");
+        data.put("label", "b");
 
         dml.setData(dataList);
 
@@ -50,8 +57,14 @@ public class LabelSyncJoinSubTest {
         Assert.assertEquals("b;a", response.getSource().get("_labels"));
     }
 
+    /**
+     * 子查询从表更新
+     */
     @Test
-    public void updateTest02() {
+    public void test02() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"update label set label='aa' where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -75,11 +88,17 @@ public class LabelSyncJoinSubTest {
         esAdapter.getEsSyncService().sync(dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
-        Assert.assertEquals("aa;b", response.getSource().get("_labels"));
+        Assert.assertEquals("b;aa", response.getSource().get("_labels"));
     }
 
+    /**
+     * 子查询从表删除
+     */
     @Test
-    public void deleteTest03() {
+    public void test03() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"delete from label where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -100,10 +119,4 @@ public class LabelSyncJoinSubTest {
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("b", response.getSource().get("_labels"));
     }
-
-    @After
-    public void after() {
-        esAdapter.destroy();
-        DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
-    }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java

@@ -14,6 +14,8 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
 import com.alibaba.otter.canal.client.adapter.support.Dml;
 
+import javax.sql.DataSource;
+
 public class RoleSyncJoinOne2Test {
 
     private ESAdapter esAdapter;
@@ -25,10 +27,14 @@ public class RoleSyncJoinOne2Test {
     }
 
     /**
-     * 非子查询从表插入 (确保主表记录必须有数据)
+     * 带函数非子查询从表插入
      */
     @Test
-    public void insertTest01() {
+    public void test01() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"delete from role where id=1");
+        Common.sqlExe(ds,"insert into role (id,role_name) values (1,'admin')");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -39,18 +45,24 @@ public class RoleSyncJoinOne2Test {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 1L);
-        data.put("role_name", "admin2");
+        data.put("role_name", "admin");
 
         dml.setData(dataList);
 
         esAdapter.getEsSyncService().sync(dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
-        Assert.assertEquals("admin2_", response.getSource().get("_role_name"));
+        Assert.assertEquals("admin_", response.getSource().get("_role_name"));
     }
 
+    /**
+     * 带函数非子查询从表更新
+     */
     @Test
-    public void updateTest02() {
+    public void test02() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"update role set role_name='admin3' where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -75,10 +87,4 @@ public class RoleSyncJoinOne2Test {
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertEquals("admin3_", response.getSource().get("_role_name"));
     }
-
-    @After
-    public void after() {
-        esAdapter.destroy();
-        DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
-    }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java

@@ -2,13 +2,13 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
 import java.util.*;
 
+import javax.sql.DataSource;
+
 import org.elasticsearch.action.get.GetResponse;
-import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
-import com.alibaba.druid.pool.DruidDataSource;
 import com.alibaba.otter.canal.client.adapter.es.ESAdapter;
 import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
 import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
@@ -25,10 +25,14 @@ public class RoleSyncJoinOneTest {
     }
 
     /**
-     * 非子查询从表插入 (确保主表记录必须有数据)
+     * 非子查询从表插入
      */
     @Test
-    public void insertTest01() {
+    public void test01() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds, "delete from role where id=1");
+        Common.sqlExe(ds, "insert into role (id,role_name) values (1,'admin')");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -49,8 +53,14 @@ public class RoleSyncJoinOneTest {
         Assert.assertEquals("admin", response.getSource().get("_role_name"));
     }
 
+    /**
+     * 非子查询从表更新
+     */
     @Test
-    public void updateTest02() {
+    public void test02() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds, "update role set role_name='admin2' where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -76,8 +86,72 @@ public class RoleSyncJoinOneTest {
         Assert.assertEquals("admin2", response.getSource().get("_role_name"));
     }
 
+    /**
+     * 主表更新外键值
+     */
     @Test
-    public void deleteTest03() {
+    public void test03() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds, "delete from role where id=2");
+        Common.sqlExe(ds, "insert into role (id,role_name) values (2,'operator')");
+        Common.sqlExe(ds, "update user set role_id=2 where id=1");
+
+         Dml dml = new Dml();
+         dml.setDestination("example");
+         dml.setTs(new Date().getTime());
+         dml.setType("UPDATE");
+         dml.setDatabase("mytest");
+         dml.setTable("user");
+         List<Map<String, Object>> dataList = new ArrayList<>();
+         Map<String, Object> data = new LinkedHashMap<>();
+         dataList.add(data);
+         data.put("id", 1L);
+         data.put("role_id", 2L);
+         dml.setData(dataList);
+         List<Map<String, Object>> oldList = new ArrayList<>();
+         Map<String, Object> old = new LinkedHashMap<>();
+         oldList.add(old);
+         old.put("role_id", 1L);
+         dml.setOld(oldList);
+         esAdapter.getEsSyncService().sync(dml);
+
+         GetResponse response =
+         esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
+         Assert.assertEquals("operator", response.getSource().get("_role_name"));
+
+        Common.sqlExe(ds, "update user set role_id=1 where id=1");
+
+        Dml dml2 = new Dml();
+        dml2.setDestination("example");
+        dml2.setTs(new Date().getTime());
+        dml2.setType("UPDATE");
+        dml2.setDatabase("mytest");
+        dml2.setTable("user");
+        List<Map<String, Object>> dataList2 = new ArrayList<>();
+        Map<String, Object> data2 = new LinkedHashMap<>();
+        dataList2.add(data2);
+        data2.put("id", 1L);
+        data2.put("role_id", 1L);
+        dml2.setData(dataList2);
+        List<Map<String, Object>> oldList2 = new ArrayList<>();
+        Map<String, Object> old2 = new LinkedHashMap<>();
+        oldList2.add(old2);
+        old2.put("role_id", 2L);
+        dml2.setOld(oldList2);
+        esAdapter.getEsSyncService().sync(dml2);
+
+        GetResponse response2 = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
+        Assert.assertEquals("admin2", response2.getSource().get("_role_name"));
+    }
+
+    /**
+     * 非子查询从表删除
+     */
+    @Test
+    public void test04() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds, "delete from role where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -97,10 +171,4 @@ public class RoleSyncJoinOneTest {
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
         Assert.assertNull(response.getSource().get("_role_name"));
     }
-
-    @After
-    public void after() {
-        esAdapter.destroy();
-        DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
-    }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java

@@ -2,6 +2,8 @@ package com.alibaba.otter.canal.client.adapter.es.test.sync;
 
 import java.util.*;
 
+import javax.sql.DataSource;
+
 import org.elasticsearch.action.get.GetResponse;
 import org.junit.After;
 import org.junit.Assert;
@@ -25,10 +27,14 @@ public class UserSyncJoinOneTest {
     }
 
     /**
-     * 主表带函数插入, 数据库里内容必须和单测一致
+     * 主表带函数插入
      */
     @Test
-    public void insertTest01() {
+    public void test01() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"delete from user where id=1");
+        Common.sqlExe(ds,"insert into user (id,name,role_id) values (1,'Eric',1)");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -51,8 +57,14 @@ public class UserSyncJoinOneTest {
         Assert.assertEquals("Eric_", response.getSource().get("_name"));
     }
 
+    /**
+     * 主表带函数更新
+     */
     @Test
-    public void updateTest02() {
+    public void test02() {
+        DataSource ds = DatasourceConfig.DATA_SOURCES.get("defaultDS");
+        Common.sqlExe(ds,"update user set name='Eric2' where id=1");
+
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -63,23 +75,17 @@ public class UserSyncJoinOneTest {
         Map<String, Object> data = new LinkedHashMap<>();
         dataList.add(data);
         data.put("id", 1L);
-        data.put("name", "Eric");
+        data.put("name", "Eric2");
         dml.setData(dataList);
         List<Map<String, Object>> oldList = new ArrayList<>();
         Map<String, Object> old = new LinkedHashMap<>();
         oldList.add(old);
-        old.put("name", "Eric2");
+        old.put("name", "Eric");
         dml.setOld(oldList);
 
         esAdapter.getEsSyncService().sync(dml);
 
         GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get();
-        Assert.assertEquals("Eric_", response.getSource().get("_name"));
-    }
-
-    @After
-    public void after() {
-        esAdapter.destroy();
-        DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
+        Assert.assertEquals("Eric2_", response.getSource().get("_name"));
     }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java

@@ -28,7 +28,7 @@ public class UserSyncSingleTest {
      * 单表插入
      */
     @Test
-    public void insertTest01() {
+    public void test01() {
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -55,7 +55,7 @@ public class UserSyncSingleTest {
      * 单表更新
      */
     @Test
-    public void updateTest02() {
+    public void test02() {
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -84,7 +84,7 @@ public class UserSyncSingleTest {
      * 单表删除
      */
     @Test
-    public void deleteTest03() {
+    public void test03() {
         Dml dml = new Dml();
         dml.setDestination("example");
         dml.setTs(new Date().getTime());
@@ -107,9 +107,9 @@ public class UserSyncSingleTest {
         Assert.assertNull(response.getSource());
     }
 
-    @After
-    public void after() {
-        esAdapter.destroy();
-        DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
-    }
+    // @After
+    // public void after() {
+    // esAdapter.destroy();
+    // DatasourceConfig.DATA_SOURCES.values().forEach(DruidDataSource::close);
+    // }
 }

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/db_schema.sql

@@ -0,0 +1,39 @@
+-- ----------------------------
+-- Table structure for label
+-- ----------------------------
+DROP TABLE IF EXISTS `label`;
+CREATE TABLE `label` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `user_id` bigint(20) NOT NULL,
+  `label` varchar(30) NOT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
+
+-- ----------------------------
+-- Table structure for role
+-- ----------------------------
+DROP TABLE IF EXISTS `role`;
+CREATE TABLE `role` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `role_name` varchar(30) NOT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
+
+-- ----------------------------
+-- Table structure for user
+-- ----------------------------
+DROP TABLE IF EXISTS `user`;
+CREATE TABLE `user` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `name` varchar(30) NOT NULL,
+  `c_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `role_id` bigint(20) DEFAULT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;
+
+insert into user (id,name,role_id) values (1,'Eric',1);
+insert into role (id,role_name) values (1,'admin');
+insert into role (id,role_name) values (2,'operator');
+insert into label (id,user_id,label) values (1,1,'a');
+insert into label (id,user_id,label) values (2,1,'b');
+commit;

client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/es_mapping.json

@@ -0,0 +1,21 @@
+{
+  "_doc": {
+    "properties": {
+      "_name": {
+        "type": "text"
+      },
+      "_role_id": {
+        "type": "long"
+      },
+      "_role_name": {
+        "type": "text"
+      },
+      "_labels": {
+        "type": "text"
+      },
+      "_c_time": {
+        "type": "date"
+      }
+    }
+  }
+}