From 62c43337908acc75824a3c42c3d0b6546da32e41 Mon Sep 17 00:00:00 2001
From: shicongyang <acongfly@163.com>
Date: Fri, 23 Dec 2022 16:04:46 +0800
Subject: [PATCH] =?UTF-8?q?canal=E4=B8=AD=E7=9A=84flatMessage=E4=B8=AD?=
 =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E4=B8=80=E4=B8=AA=E5=8F=82=E6=95=B0=EF=BC=8C?=
 =?UTF-8?q?=E8=AE=B0=E5=BD=95gtid=20#4509=20(#4521)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../canal/connector/core/producer/MQMessageUtils.java |  2 ++
 .../com/alibaba/otter/canal/protocol/FlatMessage.java | 11 ++++++++++-
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java
index 2e60ee24..1a15a8fa 100644
--- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java
+++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java
@@ -374,6 +374,7 @@ public class MQMessageUtils {
             flatMessage.setEs(entry.getHeader().getExecuteTime());
             flatMessage.setTs(System.currentTimeMillis());
             flatMessage.setSql(rowChange.getSql());
+            flatMessage.setGtid(entry.getHeader().getGtid());
 
             if (!rowChange.getIsDdl()) {
                 Map<String, Integer> sqlType = new LinkedHashMap<>();
@@ -526,6 +527,7 @@ public class MQMessageUtils {
                             flatMessageTmp.setEs(flatMessage.getEs());
                             flatMessageTmp.setTs(flatMessage.getTs());
                             flatMessageTmp.setPkNames(flatMessage.getPkNames());
+                            flatMessageTmp.setGtid(flatMessage.getGtid());
                         }
                         List<Map<String, String>> data = flatMessageTmp.getData();
                         if (data == null) {
diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
index 0957b759..90f93af1 100644
--- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
+++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
@@ -27,6 +27,7 @@ public class FlatMessage implements Serializable {
     private Map<String, String>       mysqlType;
     private List<Map<String, String>> data;
     private List<Map<String, String>> old;
+    private String                    gtid;
 
     public FlatMessage() {
     }
@@ -146,10 +147,18 @@ public class FlatMessage implements Serializable {
         this.es = es;
     }
 
+    public String getGtid() {
+        return gtid;
+    }
+
+    public void setGtid(String gtid) {
+        this.gtid = gtid;
+    }
+
     @Override
     public String toString() {
         return "FlatMessage [id=" + id + ", database=" + database + ", table=" + table + ", isDdl=" + isDdl + ", type="
                + type + ", es=" + es + ", ts=" + ts + ", sql=" + sql + ", sqlType=" + sqlType + ", mysqlType="
-               + mysqlType + ", data=" + data + ", old=" + old + "]";
+               + mysqlType + ", data=" + data + ", old=" + old + ", gtid=" + gtid +"]";
     }
 }