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

append gtid lastCommitted and sequenceNumber support

Browse Source
This commit is contained in:
winger
2018-07-18 17:07:39 +08:00
parent 8e856acac5
commit 89726a6365
2 changed files with 25 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java
View File

@@ -17,10 +17,13 @@ public class GtidLogEvent extends LogEvent {
public static final int ENCODED_FLAG_LENGTH = 1; public static final int ENCODED_FLAG_LENGTH = 1;
// / Length of SID in event encoding // / Length of SID in event encoding
public static final int ENCODED_SID_LENGTH = 16; public static final int ENCODED_SID_LENGTH = 16;
public static final int LOGICAL_TIMESTAMP_TYPE_CODE = 2;
private boolean commitFlag; private boolean commitFlag;
private UUID sid; private UUID sid;
private long gno; private long gno;
private Long lastCommitted;
private Long sequenceNumber;
public GtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ public GtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
super(header); super(header);
@@ -40,6 +43,14 @@ public class GtidLogEvent extends LogEvent {
gno = buffer.getLong64(); gno = buffer.getLong64();
// support gtid lastCommitted and sequenceNumber
// 42 = 1+16+8+1+8+8
if (buffer.capacity() > 42 && buffer.getUint8() == LOGICAL_TIMESTAMP_TYPE_CODE) {
lastCommitted = buffer.getLong64();
sequenceNumber = buffer.getLong64();
}
// ignore gtid info read // ignore gtid info read
// sid.copy_from((uchar *)ptr_buffer); // sid.copy_from((uchar *)ptr_buffer);
// ptr_buffer+= ENCODED_SID_LENGTH; // ptr_buffer+= ENCODED_SID_LENGTH;
@@ -62,4 +73,12 @@ public class GtidLogEvent extends LogEvent {
public long getGno() { public long getGno() {
return gno; return gno;
} }
public Long getLastCommitted() {
return lastCommitted;
}
public Long getSequenceNumber() {
return sequenceNumber;
}
} }

6
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
View File

@@ -167,6 +167,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
if (gtidSet != null) { if (gtidSet != null) {
gtidSet.update(value); gtidSet.update(value);
} }
if (logEvent.getLastCommitted() != null) {
builder.setKey("lastCommitted");
builder.setValue(String.valueOf(logEvent.getLastCommitted()));
builder.setKey("sequenceNumber");
builder.setValue(String.valueOf(logEvent.getSequenceNumber()));
}
Header header = createHeader(logHeader, "", "", EventType.GTID); Header header = createHeader(logHeader, "", "", EventType.GTID);
return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString()); return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());