chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

canal中的flatMessage中添加一个参数，记录gtid #4509 (#4521)

Browse Source
This commit is contained in:
shicongyang
2022-12-23 16:04:46 +08:00
committed by GitHub
parent ea8fb5e5b8
commit 62c4333790
2 changed files with 12 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java
View File

@@ -374,6 +374,7 @@ public class MQMessageUtils {
flatMessage.setEs(entry.getHeader().getExecuteTime()); flatMessage.setEs(entry.getHeader().getExecuteTime());
flatMessage.setTs(System.currentTimeMillis()); flatMessage.setTs(System.currentTimeMillis());
flatMessage.setSql(rowChange.getSql()); flatMessage.setSql(rowChange.getSql());
flatMessage.setGtid(entry.getHeader().getGtid());
if (!rowChange.getIsDdl()) { if (!rowChange.getIsDdl()) {
Map<String, Integer> sqlType = new LinkedHashMap<>(); Map<String, Integer> sqlType = new LinkedHashMap<>();
@@ -526,6 +527,7 @@ public class MQMessageUtils {
flatMessageTmp.setEs(flatMessage.getEs()); flatMessageTmp.setEs(flatMessage.getEs());
flatMessageTmp.setTs(flatMessage.getTs()); flatMessageTmp.setTs(flatMessage.getTs());
flatMessageTmp.setPkNames(flatMessage.getPkNames()); flatMessageTmp.setPkNames(flatMessage.getPkNames());
flatMessageTmp.setGtid(flatMessage.getGtid());
} }
List<Map<String, String>> data = flatMessageTmp.getData(); List<Map<String, String>> data = flatMessageTmp.getData();
if (data == null) { if (data == null) {

11
protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
View File

@@ -27,6 +27,7 @@ public class FlatMessage implements Serializable {
private Map<String, String> mysqlType; private Map<String, String> mysqlType;
private List<Map<String, String>> data; private List<Map<String, String>> data;
private List<Map<String, String>> old; private List<Map<String, String>> old;
private String gtid;
public FlatMessage() { public FlatMessage() {
} }
@@ -146,10 +147,18 @@ public class FlatMessage implements Serializable {
this.es = es; this.es = es;
} }
public String getGtid() {
return gtid;
}
public void setGtid(String gtid) {
this.gtid = gtid;
}
@Override @Override
public String toString() { public String toString() {
return "FlatMessage [id=" + id + ", database=" + database + ", table=" + table + ", isDdl=" + isDdl + ", type=" return "FlatMessage [id=" + id + ", database=" + database + ", table=" + table + ", isDdl=" + isDdl + ", type="
+ type + ", es=" + es + ", ts=" + ts + ", sql=" + sql + ", sqlType=" + sqlType + ", mysqlType=" + type + ", es=" + es + ", ts=" + ts + ", sql=" + sql + ", sqlType=" + sqlType + ", mysqlType="
+ mysqlType + ", data=" + data + ", old=" + old + "]"; + mysqlType + ", data=" + data + ", old=" + old + ", gtid=" + gtid +"]";
} }
} }