diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java index a50b2450..d0c6746d 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java @@ -92,6 +92,10 @@ public class BaseLogFetcherTest { final ColumnInfo[] columnInfo = map.getColumnInfo(); for (int i = 0; i < columnCnt; i++) { + if (!cols.get(i)) { + continue; + } + ColumnInfo info = columnInfo[i]; buffer.nextValue(info.type, info.meta); diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java index b8f4e47d..23a10f01 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java @@ -6,7 +6,6 @@ import java.sql.DriverManager; import java.sql.Statement; import org.junit.Assert; - import org.junit.Test; import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; @@ -30,7 +29,7 @@ public class DirectLogFetcherTest extends BaseLogFetcherTest { statement.execute("SET @master_binlog_checksum='@@global.binlog_checksum'"); statement.execute("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'"); - fecther.open(connection, "mysql-bin.000010", 4L, 2); + fecther.open(connection, "mysql-bin.000001", 4L, 2); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java index e7551932..7673f71b 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java @@ -5,7 +5,6 @@ import java.io.IOException; import java.net.URL; import org.junit.Assert; - import org.junit.Before; import org.junit.Test; @@ -27,7 +26,6 @@ public class FileLogFetcherTest extends BaseLogFetcherTest { URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt"); File dummyFile = new File(url.getFile()); directory = new File(dummyFile.getParent() + "/binlog").getPath(); - // directory = "/home/jianghang/tmp/binlog"; } @Test @@ -37,8 +35,8 @@ public class FileLogFetcherTest extends BaseLogFetcherTest { LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); - File current = new File(directory, "mysql-bin.000006"); - fetcher.open(current); + File current = new File(directory, "mysql-bin.000001"); + fetcher.open(current, 2051L); context.setLogPosition(new LogPosition(current.getName())); while (fetcher.fetch()) { diff --git a/deployer/src/main/bin/stop.sh b/deployer/src/main/bin/stop.sh index 73eeb9ca..f398749c 100644 --- a/deployer/src/main/bin/stop.sh +++ b/deployer/src/main/bin/stop.sh @@ -1,10 +1,14 @@ #!/bin/bash cygwin=false; +linux=false; case "`uname`" in CYGWIN*) cygwin=true ;; + Linux*) + linux=true + ;; esac get_pid() { @@ -15,11 +19,19 @@ get_pid() { JAVA_CMD=`cygpath --path --unix $JAVA_CMD` JAVA_PID=`ps |grep $JAVA_CMD |awk '{print $1}'` else - if [ ! -z "$PID" ]; then - JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` - else - JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'` - fi + if $linux; then + if [ ! -z "$PID" ]; then + JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` + else + JAVA_PID=`ps -C java -f --width 1000|grep "$STR"|grep -v grep|awk '{print $2}'` + fi + else + if [ ! -z "$PID" ]; then + JAVA_PID=`ps aux |grep "$STR"|grep "$PID"|grep -v grep|awk '{print $2}'` + else + JAVA_PID=`ps aux |grep "$STR"|grep -v grep|awk '{print $2}'` + fi + fi fi echo $JAVA_PID; } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 2831c4e1..cbcc19f8 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -41,6 +41,10 @@ canal.instance.filter.query.dml = false canal.instance.filter.query.ddl = false canal.instance.filter.table.error = false +# binlog format/image check +canal.instance.binlog.format = ROW,STATEMENT,MIXED +canal.instance.binlog.image = FULL,MINIMAL,NOBLOB + # binlog ddl isolation canal.instance.get.ddl.isolation = false diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index 1a1c3291..f12b72d3 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -179,5 +179,7 @@ + + \ No newline at end of file diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 74b48f98..618beed1 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -164,5 +164,7 @@ + + \ No newline at end of file diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index 40c7787b..c135196b 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -161,6 +161,8 @@ + + @@ -249,5 +251,7 @@ + + \ No newline at end of file diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index 524ecb3c..b7afc130 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -152,5 +152,7 @@ + + \ No newline at end of file diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 08b330a6..0133f7e1 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -27,12 +27,13 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; public class MysqlConnection implements ErosaConnection { - private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); + private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); private MysqlConnector connector; private long slaveId; - private Charset charset = Charset.forName("UTF-8"); - private BinlogFormat binlogFormat = BinlogFormat.ROW; + private Charset charset = Charset.forName("UTF-8"); + private BinlogFormat binlogFormat; + private BinlogImage binlogImage; public MysqlConnection(){ } @@ -104,11 +105,6 @@ public class MysqlConnection implements ErosaConnection { public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); - BinlogFormat format = getBinlogFormat(); - if (!format.isRow()) { - logger.warn("binlog_format : {} , it will not have rowData before/after columns", format.toString()); - } - sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); @@ -212,7 +208,7 @@ public class MysqlConnection implements ErosaConnection { } /** - * 判断一下是否采用ROW模式 + * 获取一下binlog format格式 */ private void loadBinlogFormat() { ResultSetPacket rs = null; @@ -234,6 +230,30 @@ public class MysqlConnection implements ErosaConnection { } } + /** + * 获取一下binlog image格式 + */ + private void loadBinlogImage() { + ResultSetPacket rs = null; + try { + rs = query("show variables like 'binlog_row_image'"); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues == null || columnValues.size() != 2) { + // 可能历时版本没有image特性 + binlogImage = BinlogImage.FULL; + } else { + binlogImage = BinlogImage.valuesOf(columnValues.get(1)); + } + + if (binlogFormat == null) { + throw new IllegalStateException("unexpected binlog image query result:" + rs.getFieldValues()); + } + } + public static enum BinlogFormat { STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED"); @@ -267,6 +287,46 @@ public class MysqlConnection implements ErosaConnection { } } + /** + * http://dev.mysql.com/doc/refman/5.6/en/replication-options-binary-log. + * html#sysvar_binlog_row_image + * + * @author agapple 2015年6月29日 下午10:39:03 + * @since 1.0.20 + */ + public static enum BinlogImage { + + FULL("FULL"), MINIMAL("MINIMAL"), NOBLOB("NOBLOB"); + + public boolean isFull() { + return this == FULL; + } + + public boolean isMinimal() { + return this == MINIMAL; + } + + public boolean isNoBlob() { + return this == NOBLOB; + } + + private String value; + + private BinlogImage(String value){ + this.value = value; + } + + public static BinlogImage valuesOf(String value) { + BinlogImage[] formats = values(); + for (BinlogImage format : formats) { + if (format.value.equalsIgnoreCase(value)) { + return format; + } + } + return null; + } + } + // ================== setter / getter =================== public Charset getCharset() { @@ -303,4 +363,14 @@ public class MysqlConnection implements ErosaConnection { return binlogFormat; } + public BinlogImage getBinlogImage() { + if (binlogImage == null) { + synchronized (this) { + loadBinlogImage(); + } + } + + return binlogImage; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 787b72c9..b779e03c 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -22,6 +22,8 @@ import com.alibaba.otter.canal.parse.ha.CanalHAController; import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.HeartBeatCallback; import com.alibaba.otter.canal.parse.inbound.SinkFunction; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.BinlogFormat; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.BinlogImage; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; @@ -62,6 +64,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE private TableMetaCache tableMetaCache; // 对应meta // cache private int fallbackIntervalInSeconds = 60; // 切换回退时间 + private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验 + private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验 // 心跳检查 @@ -82,6 +86,34 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE throw new CanalParseException(e); } + if (supportBinlogFormats != null && supportBinlogFormats.length > 0) { + BinlogFormat format = ((MysqlConnection) metaConnection).getBinlogFormat(); + boolean found = false; + for (BinlogFormat supportFormat : supportBinlogFormats) { + if (supportFormat != null && format == supportFormat) { + found = true; + } + } + + if (!found) { + throw new CanalParseException("Unsupported BinlogFormat " + format); + } + } + + if (supportBinlogImages != null && supportBinlogImages.length > 0) { + BinlogImage image = ((MysqlConnection) metaConnection).getBinlogImage(); + boolean found = false; + for (BinlogImage supportImage : supportBinlogImages) { + if (supportImage != null && image == supportImage) { + found = true; + } + } + + if (!found) { + throw new CanalParseException("Unsupported BinlogImage " + image); + } + } + tableMetaCache = new TableMetaCache(metaConnection); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); } @@ -685,6 +717,32 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE return event; } + public void setSupportBinlogFormats(String formatStrs) { + String[] formats = StringUtils.split(formatStrs, ','); + if (formats != null) { + BinlogFormat[] supportBinlogFormats = new BinlogFormat[formats.length]; + int i = 0; + for (String format : formats) { + supportBinlogFormats[i++] = BinlogFormat.valuesOf(format); + } + + this.supportBinlogFormats = supportBinlogFormats; + } + } + + public void setSupportBinlogImages(String imageStrs) { + String[] images = StringUtils.split(imageStrs, ','); + if (images != null) { + BinlogImage[] supportBinlogImages = new BinlogImage[images.length]; + int i = 0; + for (String image : images) { + supportBinlogImages[i++] = BinlogImage.valuesOf(image); + } + + this.supportBinlogImages = supportBinlogImages; + } + } + // ===================== setter / getter ======================== public void setDefaultConnectionTimeoutInSeconds(int defaultConnectionTimeoutInSeconds) { 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 14fd3498..c3465e8a 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 @@ -422,6 +422,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar for (int i = 0; i < columnCnt; i++) { ColumnInfo info = columnInfo[i]; + // mysql 5.6开始支持nolob/mininal类型,并不一定记录所有的列,需要进行判断 + if (!cols.get(i)) { + continue; + } + Column.Builder columnBuilder = Column.newBuilder(); FieldMeta fieldMeta = null; @@ -604,25 +609,21 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar if (index < 0) { return false; } - if ((bfColumns.size() - 1) < index) { - return false; - } - Column column = bfColumns.get(index); - if (column.getIsNull()) { - if (newValue != null) { - return true; - } - } else { - if (newValue == null) { - return true; - } else { - if (!column.getValue().equals(newValue)) { - return true; + for (Column column : bfColumns) { + if (column.getIndex() == index) {// 比较before / after的column index + if (column.getIsNull() && newValue == null) { + // 如果全是null + return false; + } else if (newValue != null && column.getValue().equals(newValue)) { + // 如果不围null，并且相等 + return false; } } } - return false; + + // 比如nolob/minial模式下,可能找不到before记录,认为是有变化 + return true; } private TableMeta getTableMeta(String dbName, String tbName, boolean useCache) {