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

修复开启checksum后rotateLogEvent数据包解析时binlog name乱码

Browse Source
This commit is contained in:
李建安
2016-10-10 18:30:09 +08:00
parent 0c9eecd758
commit 96454d4387
4 changed files with 33 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java
View File

@@ -141,6 +141,11 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 {
}
}
public FormatDescriptionLogEvent(final int binlogVersion,int binlogChecksum){
this(binlogVersion);
this.header.checksumAlg = binlogChecksum;
}
public FormatDescriptionLogEvent(final int binlogVersion){
this.binlogVersion = binlogVersion;

1
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RotateLogEvent.java
View File

@@ -90,6 +90,7 @@ public final class RotateLogEvent extends LogEvent {
int filenameLen = buffer.limit() - filenameOffset;
if (filenameLen > FN_REFLEN - 1) filenameLen = FN_REFLEN - 1;
buffer.position(filenameOffset);
filename = buffer.getFixString(filenameLen);
}

1
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
View File

@@ -325,7 +325,6 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
startTs = System.currentTimeMillis();
}
CanalEntry.Entry event = binlogParser.parse(bod);
if (enabled) {
this.parsingInterval = System.currentTimeMillis() - startTs;
}

24
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -7,6 +7,8 @@ import java.nio.charset.Charset;
import java.util.List;
import com.taobao.tddl.dbsync.binlog.LogPosition;
import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,6 +37,7 @@ public class MysqlConnection implements ErosaConnection {
private Charset charset = Charset.forName("UTF-8");
private BinlogFormat binlogFormat;
private BinlogImage binlogImage;
private int binlogChecksum;
public MysqlConnection(){
}
@@ -107,12 +110,14 @@ public class MysqlConnection implements ErosaConnection {
public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
updateSettings();
loadBinlogChecksum();
sendBinlogDump(binlogfilename, binlogPosition);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
fetcher.start(connector.getChannel());
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
context.setLogPosition(new LogPosition(binlogfilename));
context.setFormatDescription(new FormatDescriptionLogEvent(4,binlogChecksum));
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
@@ -259,6 +264,25 @@ public class MysqlConnection implements ErosaConnection {
throw new IllegalStateException("unexpected binlog image query result:" + rs.getFieldValues());
}
}
/**
* 获取主库checksum信息
* https://dev.mysql.com/doc/refman/5.6/en/replication-options-binary-log.html#option_mysqld_binlog-checksum
*/
private void loadBinlogChecksum(){
ResultSetPacket rs = null;
try {
rs = query("select @master_binlog_checksum");
} catch (IOException e) {
throw new CanalParseException(e);
}
List<String> columnValues = rs.getFieldValues();
if(columnValues.get(0).toUpperCase().equals("CRC32")){
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
}else{
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
}
}
public static enum BinlogFormat {