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 dbb19701..0ce61df3 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 @@ -418,6 +418,12 @@ public final class LogDecoder { logPosition.position = header.getLogPos(); return event; } + case LogEvent.HEARTBEAT_LOG_EVENT_V2: { + HeartbeatV2LogEvent event = new HeartbeatV2LogEvent(header, buffer, descriptionEvent); + /* updating position in context */ + logPosition.position = header.getLogPos(); + return event; + } default: /* * Create an object of Ignorable_log_event for unrecognized 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 74e83c33..c200d3d9 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 @@ -178,7 +178,10 @@ public abstract class LogEvent { /* mysql 8.0.20 */ public static final int TRANSACTION_PAYLOAD_EVENT = 40; - public static final int MYSQL_ENUM_END_EVENT = 41; + /* mysql 8.0.26 */ + public static final int HEARTBEAT_LOG_EVENT_V2 = 41; + + public static final int MYSQL_ENUM_END_EVENT = 42; // mariaDb 5.5.34 /* New MySQL/Sun events are to be added right above this comment */ @@ -358,6 +361,7 @@ public abstract class LogEvent { case INCIDENT_EVENT: return "Incident"; case HEARTBEAT_LOG_EVENT: + case HEARTBEAT_LOG_EVENT_V2: return "Heartbeat"; case IGNORABLE_LOG_EVENT: return "Ignorable"; diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java new file mode 100644 index 00000000..e6619c7c --- /dev/null +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java @@ -0,0 +1,40 @@ +package com.taobao.tddl.dbsync.binlog.event; + +import com.taobao.tddl.dbsync.binlog.LogBuffer; +import com.taobao.tddl.dbsync.binlog.LogEvent; + +/** + *
+ * Replication event to ensure to replica that source is alive. + * The event is originated by source's dump thread and sent straight to + * replica without being logged. Slave itself does not store it in relay log + * but rather uses a data for immediate checks and throws away the event. + * Two members of the class m_log_filename and m_log_position comprise + * @see the rpl_event_coordinates instance. The coordinates that a heartbeat + * instance carries correspond to the last event source has sent from + * its binlog. + * Also this event will be generated only for the source server with + * version > 8.0.26 + *+ * + * @author jianghang 2022-09-01 下午16:36:29 + * @version 1.1.6 + * @since mysql 8.0.26 + */ +public class HeartbeatV2LogEvent extends LogEvent { + + private byte[] payload; + + public HeartbeatV2LogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ + super(header); + + final int commonHeaderLen = descriptionEvent.commonHeaderLen; + int payloadLenth = buffer.limit() - commonHeaderLen; + // see : https://github.com/mysql/mysql-server/commit/59e590738a772b74ad50c4a57c86aaa1bc6501c7#diff-184e9a7d8a58f080974e475d4199fe5c6da5518c8a2811cc5df5988c8f9e9797 + payload = buffer.getData(payloadLenth); + } + + public byte[] getPayload() { + return payload; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 8636acfd..10f62a4f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -12,6 +12,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import com.taobao.tddl.dbsync.binlog.event.*; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; @@ -40,23 +41,7 @@ import com.alibaba.otter.canal.protocol.CanalEntry.Type; import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.google.protobuf.ByteString; import com.taobao.tddl.dbsync.binlog.LogEvent; -import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent; -import com.taobao.tddl.dbsync.binlog.event.HeartbeatLogEvent; -import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent; -import com.taobao.tddl.dbsync.binlog.event.LogHeader; -import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RandLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer; -import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; -import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent; -import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo; -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.WriteRowsLogEvent; -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.MariaGtidListLogEvent; import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidLogEvent; @@ -147,6 +132,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return parseGTIDLogEvent((GtidLogEvent) logEvent); case LogEvent.HEARTBEAT_LOG_EVENT: return parseHeartbeatLogEvent((HeartbeatLogEvent) logEvent); + case LogEvent.HEARTBEAT_LOG_EVENT_V2: + return parseHeartbeatV2LogEvent((HeartbeatV2LogEvent) logEvent); case LogEvent.GTID_EVENT: case LogEvent.GTID_LIST_EVENT: return parseMariaGTIDLogEvent(logEvent); @@ -173,6 +160,15 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return entryBuilder.build(); } + private Entry parseHeartbeatV2LogEvent(HeartbeatV2LogEvent logEvent) { + Header.Builder headerBuilder = Header.newBuilder(); + headerBuilder.setEventType(EventType.MHEARTBEAT); + Entry.Builder entryBuilder = Entry.newBuilder(); + entryBuilder.setHeader(headerBuilder.build()); + entryBuilder.setEntryType(EntryType.HEARTBEAT); + return entryBuilder.build(); + } + private Entry parseGTIDLogEvent(GtidLogEvent logEvent) { LogHeader logHeader = logEvent.getHeader(); Pair.Builder builder = Pair.newBuilder();