[ISSUE #5040] Support gtid mode for sync data with mysql (#5041)

* [ISSUE #5040] Support gtid mode for sync data with mysql

* fix conflicts with master

* fix checkstyle error

Author: xwm1992

eventmesh-admin-server/conf/eventmesh.sql

@@ -71,8 +71,11 @@ CREATE TABLE IF NOT EXISTS `event_mesh_job_info` (
 CREATE TABLE IF NOT EXISTS `event_mesh_mysql_position` (
   `id` int unsigned NOT NULL AUTO_INCREMENT,
   `jobID` int unsigned NOT NULL,
+  `serverUUID` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
   `address` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
   `position` bigint DEFAULT NULL,
+  `gtid` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
+  `currentGtid` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
   `timestamp` bigint DEFAULT NULL,
   `journalName` varchar(50) COLLATE utf8mb4_general_ci DEFAULT NULL,
   `createTime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,

eventmesh-admin-server/conf/mapper/EventMeshMysqlPositionMapper.xml

@@ -16,24 +16,28 @@
   limitations under the License.
   -->
 <!DOCTYPE mapper
-        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
-        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+    PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
+    "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="org.apache.eventmesh.admin.server.web.db.mapper.EventMeshMysqlPositionMapper">
 
     <resultMap id="BaseResultMap" type="org.apache.eventmesh.admin.server.web.db.entity.EventMeshMysqlPosition">
-            <id property="id" column="id" jdbcType="INTEGER"/>
-            <result property="jobID" column="jobID" jdbcType="INTEGER"/>
-            <result property="address" column="address" jdbcType="VARCHAR"/>
-            <result property="position" column="position" jdbcType="BIGINT"/>
-            <result property="timestamp" column="timestamp" jdbcType="BIGINT"/>
-            <result property="journalName" column="journalName" jdbcType="VARCHAR"/>
-            <result property="createTime" column="createTime" jdbcType="TIMESTAMP"/>
-            <result property="updateTime" column="updateTime" jdbcType="TIMESTAMP"/>
+        <id property="id" column="id" jdbcType="INTEGER"/>
+        <result property="jobID" column="jobID" jdbcType="INTEGER"/>
+        <result property="serverUUID" column="serverUUID" jdbcType="VARCHAR"/>
+        <result property="address" column="address" jdbcType="VARCHAR"/>
+        <result property="position" column="position" jdbcType="BIGINT"/>
+        <result property="gtid" column="gtid" jdbcType="VARCHAR"/>
+        <result property="currentGtid" column="currentGtid" jdbcType="VARCHAR"/>
+        <result property="timestamp" column="timestamp" jdbcType="BIGINT"/>
+        <result property="journalName" column="journalName" jdbcType="VARCHAR"/>
+        <result property="createTime" column="createTime" jdbcType="TIMESTAMP"/>
+        <result property="updateTime" column="updateTime" jdbcType="TIMESTAMP"/>
     </resultMap>
 
     <sql id="Base_Column_List">
-        id,jobID,address,
-        position,timestamp,journalName,
+        id
+        ,jobID,serverUUID,address,
+        position,gtid,currentGtid,timestamp,journalName,
         createTime,updateTime
     </sql>
 </mapper>

eventmesh-admin-server/src/main/java/org/apache/eventmesh/admin/server/web/db/entity/EventMeshMysqlPosition.java

@@ -38,10 +38,16 @@ public class EventMeshMysqlPosition implements Serializable {
 
     private Integer jobID;
 
+    private String serverUUID;
+
     private String address;
 
     private Long position;
 
+    private String gtid;
+
+    private String currentGtid;
+
     private Long timestamp;
 
     private String journalName;

eventmesh-admin-server/src/main/java/org/apache/eventmesh/admin/server/web/service/position/impl/MysqlPositionHandler.java

@@ -115,9 +115,12 @@ public boolean handler(ReportPositionRequest request, Metadata metadata) {
                 CanalRecordOffset offset = (CanalRecordOffset) recordPosition.getRecordOffset();
                 if (offset != null) {
                     position.setPosition(offset.getOffset());
+                    position.setGtid(offset.getGtid());
+                    position.setCurrentGtid(offset.getCurrentGtid());
                 }
                 CanalRecordPartition partition = (CanalRecordPartition) recordPosition.getRecordPartition();
                 if (partition != null) {
+                    position.setServerUUID(partition.getServerUUID());
                     position.setTimestamp(partition.getTimeStamp());
                     position.setJournalName(partition.getJournalName());
                 }
@@ -148,13 +151,16 @@ public List<RecordPosition> handler(FetchPositionRequest request, Metadata metad
             request.getJobID()));
         List<RecordPosition> recordPositionList = new ArrayList<>();
         for (EventMeshMysqlPosition position : positionList) {
-            RecordPosition recordPosition = new RecordPosition();
             CanalRecordPartition partition = new CanalRecordPartition();
             partition.setTimeStamp(position.getTimestamp());
             partition.setJournalName(position.getJournalName());
+            partition.setServerUUID(position.getServerUUID());
+            RecordPosition recordPosition = new RecordPosition();
             recordPosition.setRecordPartition(partition);
             CanalRecordOffset offset = new CanalRecordOffset();
             offset.setOffset(position.getPosition());
+            offset.setGtid(position.getGtid());
+            offset.setCurrentGtid(position.getCurrentGtid());
             recordPosition.setRecordOffset(offset);
             recordPositionList.add(recordPosition);
         }

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSinkConfig.java

@@ -39,6 +39,8 @@ public class CanalSinkConfig extends SinkConfig {
     // sync mode: field/row
     private SyncMode syncMode;
 
+    private boolean isGTIDMode = true;
+
     // skip sink process exception
     private Boolean skipException = false;
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/config/connector/rdb/canal/CanalSourceConfig.java

@@ -45,6 +45,10 @@ public class CanalSourceConfig extends SourceConfig {
 
     private Short clientId;
 
+    private String serverUUID;
+
+    private boolean isGTIDMode = true;
+
     private Integer batchSize = 10000;
 
     private Long batchTimeout = -1L;

eventmesh-common/src/main/java/org/apache/eventmesh/common/remote/offset/canal/CanalRecordOffset.java

@@ -30,6 +30,11 @@ public class CanalRecordOffset extends RecordOffset {
 
     private Long offset;
 
+    // mysql instance gtid range
+    private String gtid;
+
+    private String currentGtid;
+
     @Override
     public Class<? extends RecordOffset> getRecordOffsetClass() {
         return CanalRecordOffset.class;

eventmesh-common/src/main/java/org/apache/eventmesh/common/remote/offset/canal/CanalRecordPartition.java

@@ -29,6 +29,8 @@
 @ToString
 public class CanalRecordPartition extends RecordPartition {
 
+    private String serverUUID;
+
     private String journalName;
 
     private Long timeStamp;

eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/CanalConnectRecord.java

@@ -31,8 +31,14 @@
 public class CanalConnectRecord {
 
     private String schemaName;
+
     private String tableName;
 
+    // mysql instance gtid range
+    private String gtid;
+
+    private String currentGtid;
+
     /**
      * The business type of the changed data (I/U/D/C/A/E), consistent with the EventType defined in EntryProtocol in canal.
      */

eventmesh-connectors/eventmesh-connector-canal/src/main/java/org/apache/eventmesh/connector/canal/dialect/AbstractDbDialect.java

@@ -97,10 +97,6 @@ public SqlTemplate getSqlTemplate() {
         return sqlTemplate;
     }
 
-    public boolean isDRDS() {
-        return false;
-    }
-
     public String getShardColumns(String schema, String table) {
         return null;
     }