From 99aefaa1188b8bc004722ee9ce1d8f58203b0a76 Mon Sep 17 00:00:00 2001 From: "jianghang.loujh" Date: Mon, 23 May 2022 23:23:52 +0800 Subject: [PATCH] support mysql 8.0 transaction_payload_event --- .../src/main/resources/application.yml | 2 +- .../taobao/tddl/dbsync/binlog/LogDecoder.java | 41 ++++--------------- .../taobao/tddl/dbsync/binlog/LogEvent.java | 15 ++++++- .../event/TransactionPayloadLogEvent.java | 16 ++++++++ 4 files changed, 39 insertions(+), 35 deletions(-) create mode 100644 dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TransactionPayloadLogEvent.java diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 86de28ec..5b8317cc 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -11,7 +11,7 @@ canal.conf: flatMessage: true zookeeperHosts: syncBatchSize: 1000 - retries: 0 + retries: -1 timeout: accessKey: secretKey: diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java index 989a34e0..b95e4aba 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java @@ -3,42 +3,11 @@ package com.taobao.tddl.dbsync.binlog; import java.io.IOException; import java.util.BitSet; +import com.taobao.tddl.dbsync.binlog.event.*; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; -import com.taobao.tddl.dbsync.binlog.event.AppendBlockLogEvent; -import com.taobao.tddl.dbsync.binlog.event.BeginLoadQueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.CreateFileLogEvent; -import com.taobao.tddl.dbsync.binlog.event.DeleteFileLogEvent; -import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.ExecuteLoadLogEvent; -import com.taobao.tddl.dbsync.binlog.event.ExecuteLoadQueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; -import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent; -import com.taobao.tddl.dbsync.binlog.event.HeartbeatLogEvent; -import com.taobao.tddl.dbsync.binlog.event.IgnorableLogEvent; -import com.taobao.tddl.dbsync.binlog.event.IncidentLogEvent; -import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent; -import com.taobao.tddl.dbsync.binlog.event.LoadLogEvent; -import com.taobao.tddl.dbsync.binlog.event.LogHeader; -import com.taobao.tddl.dbsync.binlog.event.PreviousGtidsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RandLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.StartLogEventV3; -import com.taobao.tddl.dbsync.binlog.event.StopLogEvent; -import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; -import com.taobao.tddl.dbsync.binlog.event.TransactionContextLogEvent; -import com.taobao.tddl.dbsync.binlog.event.UnknownLogEvent; -import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.UserVarLogEvent; -import com.taobao.tddl.dbsync.binlog.event.ViewChangeEvent; -import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.XaPrepareLogEvent; -import com.taobao.tddl.dbsync.binlog.event.XidLogEvent; import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent; import com.taobao.tddl.dbsync.binlog.event.mariadb.BinlogCheckPointLogEvent; import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidListLogEvent; @@ -386,6 +355,12 @@ public final class LogDecoder { logPosition.position = header.getLogPos(); return event; } + case LogEvent.TRANSACTION_PAYLOAD_EVENT: { + TransactionPayloadLogEvent event = new TransactionPayloadLogEvent(header, buffer, descriptionEvent); + /* updating position in context */ + logPosition.position = header.getLogPos(); + return event; + } case LogEvent.VIEW_CHANGE_EVENT: { ViewChangeEvent event = new ViewChangeEvent(header, buffer, descriptionEvent); /* updating position in context */ @@ -443,7 +418,7 @@ public final class LogDecoder { logPosition.position = header.getLogPos(); return event; } - default: + com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService.lambda default: /* * Create an object of Ignorable_log_event for unrecognized * sub-class. So that SLAVE SQL THREAD will only update the diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java index 7c3e33de..74e83c33 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java @@ -175,6 +175,11 @@ public abstract class LogEvent { */ public static final int PARTIAL_UPDATE_ROWS_EVENT = 39; + /* mysql 8.0.20 */ + public static final int TRANSACTION_PAYLOAD_EVENT = 40; + + public static final int MYSQL_ENUM_END_EVENT = 41; + // mariaDb 5.5.34 /* New MySQL/Sun events are to be added right above this comment */ public static final int MYSQL_EVENTS_END = 49; @@ -372,8 +377,16 @@ public abstract class LogEvent { return "Previous_gtids"; case PARTIAL_UPDATE_ROWS_EVENT: return "Update_rows_partial"; + case TRANSACTION_CONTEXT_EVENT : + return "Transaction_context"; + case VIEW_CHANGE_EVENT : + return "view_change"; + case XA_PREPARE_LOG_EVENT : + return "Xa_prepare"; + case TRANSACTION_PAYLOAD_EVENT : + return "transaction_payload"; default: - return "Unknown"; /* impossible */ + return "Unknown type:" + type; } } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TransactionPayloadLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TransactionPayloadLogEvent.java new file mode 100644 index 00000000..95695ab5 --- /dev/null +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TransactionPayloadLogEvent.java @@ -0,0 +1,16 @@ +package com.taobao.tddl.dbsync.binlog.event; + +import com.taobao.tddl.dbsync.binlog.LogBuffer; +import com.taobao.tddl.dbsync.binlog.LogEvent; + +/** + * @author agapple 2022年5月23日 下午7:05:39 + * @version 1.1.6 + * @since mysql 8.0.20 + */ +public class TransactionPayloadLogEvent extends LogEvent { + + public TransactionPayloadLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ + super(header); + } +}