diff --git a/client-launcher/src/main/bin/startup.sh b/client-launcher/src/main/bin/startup.sh index d467b4d2..0e09ddff 100644 --- a/client-launcher/src/main/bin/startup.sh +++ b/client-launcher/src/main/bin/startup.sh @@ -74,7 +74,7 @@ else fi JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8" -CANAL_OPTS="-DappName=otter-canal -Dlogback.configurationFile=$logback_configurationFile -Dcanal.conf=$canal_conf" +CANAL_OPTS="-DappName=otter-canal-client -Dlogback.configurationFile=$logback_configurationFile -Dcanal.conf=$canal_conf" if [ -e $canal_conf -a -e $logback_configurationFile ] then diff --git a/client-launcher/src/main/bin/stop.sh b/client-launcher/src/main/bin/stop.sh index 0cfcc712..c96f2807 100644 --- a/client-launcher/src/main/bin/stop.sh +++ b/client-launcher/src/main/bin/stop.sh @@ -45,7 +45,7 @@ fi pid=`cat $pidfile` if [ "$pid" == "" ] ; then - pid=`get_pid "appName=otter-canal"` + pid=`get_pid "appName=otter-canal-client"` fi echo -e "`hostname`: stopping canal $pid ... " @@ -54,7 +54,7 @@ kill $pid LOOPS=0 while (true); do - gpid=`get_pid "appName=otter-canal" "$pid"` + gpid=`get_pid "appName=otter-canal-client" "$pid"` if [ "$gpid" == "" ] ; then echo "Oook! cost:$LOOPS" `rm $pidfile` diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java index 80b2c254..d8699629 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogContext.java @@ -4,9 +4,7 @@ import java.util.HashMap; import java.util.Map; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; -import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; -import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent; import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; /** @@ -66,18 +64,11 @@ public final class LogContext { mapOfTable.clear(); } - public final void putGtid(GtidLogEvent logEvent) { - if (logEvent != null) { - String gtid = logEvent.getSid().toString() + ":" + logEvent.getGno(); - if (gtidSet == null) { - gtid = logEvent.getSid().toString() + ":1-" + logEvent.getGno(); - gtidSet = MysqlGTIDSet.parse(gtid); - } - gtidSet.update(gtid); - } - } - public GTIDSet getGtidSet() { return gtidSet; } + + public void setGtidSet(GTIDSet gtidSet) { + this.gtidSet = gtidSet; + } } 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 d094b2fd..cfe51a1c 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,10 +3,10 @@ package com.taobao.tddl.dbsync.binlog; import java.io.IOException; import java.util.BitSet; -import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; 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; @@ -47,7 +47,7 @@ import com.taobao.tddl.dbsync.binlog.event.mariadb.StartEncryptionLogEvent; /** * Implements a binary-log decoder. - * + * * 
  * LogDecoder decoder = new LogDecoder();
  * decoder.handle(...);
@@ -62,7 +62,7 @@ import com.taobao.tddl.dbsync.binlog.event.mariadb.StartEncryptionLogEvent;
  * while (event != null);
  * // no more events in buffer.
  *
- * + * * @author Changyuan.lh * @version 1.0 */ @@ -89,7 +89,7 @@ public final class LogDecoder { /** * Decoding an event from binary-log buffer. - * + * * @return UknownLogEvent if event type is unknown or skipped, * null if buffer is not including a full event. */ @@ -141,7 +141,7 @@ public final class LogDecoder { /** * Deserialize an event from buffer. - * + * * @return UknownLogEvent if event type is unknown or skipped. */ public static LogEvent decode(LogBuffer buffer, LogHeader header, LogContext context) throws IOException { @@ -160,6 +160,7 @@ public final class LogDecoder { // remove checksum bytes buffer.limit(header.getEventLen() - LogEvent.BINLOG_CHECKSUM_LEN); } + GTIDSet gtidSet = context.getGtidSet(); switch (header.getType()) { case LogEvent.QUERY_EVENT: { QueryLogEvent event = new QueryLogEvent(header, buffer, descriptionEvent); @@ -373,8 +374,11 @@ public final class LogDecoder { GtidLogEvent event = new GtidLogEvent(header, buffer, descriptionEvent); /* updating position in context */ logPosition.position = header.getLogPos(); - // update latest gtid - context.putGtid(event); + if (gtidSet != null) { + gtidSet.update(event.getGtidStr()); + // update latest gtid + header.putGtidStr(gtidSet); + } return event; } case LogEvent.PREVIOUS_GTIDS_LOG_EVENT: { diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java index 5c2678b5..49557fc6 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java @@ -14,10 +14,10 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; public class GtidLogEvent extends LogEvent { // / Length of the commit_flag in event encoding - public static final int ENCODED_FLAG_LENGTH = 1; + public static final int ENCODED_FLAG_LENGTH = 1; // / Length of SID in event encoding - public static final int ENCODED_SID_LENGTH = 16; - public static final int LOGICAL_TIMESTAMP_TYPE_CODE = 2; + public static final int ENCODED_SID_LENGTH = 16; + public static final int LOGICAL_TIMESTAMP_TYPE_CODE = 2; private boolean commitFlag; private UUID sid; @@ -50,8 +50,6 @@ public class GtidLogEvent extends LogEvent { sequenceNumber = buffer.getLong64(); } - - // ignore gtid info read // sid.copy_from((uchar *)ptr_buffer); // ptr_buffer+= ENCODED_SID_LENGTH; @@ -82,4 +80,11 @@ public class GtidLogEvent extends LogEvent { public Long getSequenceNumber() { return sequenceNumber; } + + public String getGtidStr() { + StringBuilder sb = new StringBuilder(); + sb.append(sid.toString()).append(":"); + sb.append(gno); + return sb.toString(); + } } diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java index f5af66e3..d9b6355b 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java @@ -15,9 +15,7 @@ public class FetcherPerformanceTest { DirectLogFetcher fetcher = new DirectLogFetcher(); try { Class.forName("com.mysql.jdbc.Driver"); - Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306", - "root", - "hello"); + Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306", "root", "hello"); Statement statement = connection.createStatement(); statement.execute("SET @master_binlog_checksum='@@global.binlog_checksum'"); statement.execute("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'"); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index aacace84..42821405 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -387,7 +387,7 @@ public class CanalController { return config; } - private String getProperty(Properties properties, String key) { + public String getProperty(Properties properties, String key) { key = StringUtils.trim(key); String value = System.getProperty(key); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 2524186a..62b73b1d 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -3,12 +3,13 @@ package com.alibaba.otter.canal.deployer; import java.io.FileInputStream; import java.util.Properties; -import com.alibaba.otter.canal.kafka.CanalKafkaStarter; -import com.alibaba.otter.canal.server.CanalServerStarter; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.kafka.CanalKafkaStarter; +import com.alibaba.otter.canal.server.CanalServerStarter; + /** * canal独立版本启动的入口类 * @@ -55,12 +56,13 @@ public class CanalLauncher { }); CanalServerStarter canalServerStarter = null; - String serverMode = properties.getProperty(CanalConstants.CANAL_SERVER_MODE, "tcp"); + String serverMode = controller.getProperty(properties, CanalConstants.CANAL_SERVER_MODE); if (serverMode.equalsIgnoreCase("kafka")) { canalServerStarter = new CanalKafkaStarter(); } else if (serverMode.equalsIgnoreCase("rocketMQ")) { // 预留rocketMQ启动 } + if (canalServerStarter != null) { canalServerStarter.init(); } diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index edd6b464..db4a9abe 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,6 +12,4 @@ canalGetTimeout: 100 canalDestinations: - canalDestination: example topic: example - partition: - - + partition: \ No newline at end of file diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java index f746189e..8de2acfd 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java @@ -138,6 +138,16 @@ public class BioSocketChannel implements SocketChannel { if (socket != null) { return socket.getRemoteSocketAddress(); } + + return null; + } + + public SocketAddress getLocalSocketAddress() { + Socket socket = this.socket; + if (socket != null) { + return socket.getLocalSocketAddress(); + } + return null; } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java index 782af21e..878dc605 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java @@ -216,6 +216,10 @@ public class NettySocketChannel implements SocketChannel { return channel != null ? channel.remoteAddress() : null; } + public SocketAddress getLocalSocketAddress() { + return channel != null ? channel.localAddress() : null; + } + public void close() { if (channel != null) { channel.close(); diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java index 67f967ab..b67702ae 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java @@ -21,5 +21,7 @@ public interface SocketChannel { public SocketAddress getRemoteSocketAddress(); + public SocketAddress getLocalSocketAddress(); + public void close(); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/PositionNotFoundException.java b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/PositionNotFoundException.java index d5552a9f..67a1d72d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/PositionNotFoundException.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/PositionNotFoundException.java @@ -8,23 +8,23 @@ public class PositionNotFoundException extends CanalParseException { private static final long serialVersionUID = -7382448928116244017L; - public PositionNotFoundException(String errorCode) { + public PositionNotFoundException(String errorCode){ super(errorCode); } - public PositionNotFoundException(String errorCode, Throwable cause) { + public PositionNotFoundException(String errorCode, Throwable cause){ super(errorCode, cause); } - public PositionNotFoundException(String errorCode, String errorDesc) { + public PositionNotFoundException(String errorCode, String errorDesc){ super(errorCode, errorDesc); } - public PositionNotFoundException(String errorCode, String errorDesc, Throwable cause) { + public PositionNotFoundException(String errorCode, String errorDesc, Throwable cause){ super(errorCode, errorDesc, cause); } - public PositionNotFoundException(Throwable cause) { + public PositionNotFoundException(Throwable cause){ super(cause); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java index 05baee02..4186c6ec 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/exception/ServerIdNotMatchException.java @@ -6,27 +6,27 @@ import com.alibaba.otter.canal.common.CanalException; * @author chengjin.lyf on 2018/8/8 下午1:07 * @since 1.0.25 */ -public class ServerIdNotMatchException extends CanalException{ +public class ServerIdNotMatchException extends CanalException { private static final long serialVersionUID = -6124989280379293953L; - public ServerIdNotMatchException(String errorCode) { + public ServerIdNotMatchException(String errorCode){ super(errorCode); } - public ServerIdNotMatchException(String errorCode, Throwable cause) { + public ServerIdNotMatchException(String errorCode, Throwable cause){ super(errorCode, cause); } - public ServerIdNotMatchException(String errorCode, String errorDesc) { + public ServerIdNotMatchException(String errorCode, String errorDesc){ super(errorCode, errorDesc); } - public ServerIdNotMatchException(String errorCode, String errorDesc, Throwable cause) { + public ServerIdNotMatchException(String errorCode, String errorDesc, Throwable cause){ super(errorCode, errorDesc, cause); } - public ServerIdNotMatchException(Throwable cause) { + public ServerIdNotMatchException(Throwable cause){ super(cause); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index fcae1317..e1f80d76 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -19,11 +19,13 @@ import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler; import com.alibaba.otter.canal.filter.CanalEventFilter; import com.alibaba.otter.canal.parse.CanalEventParser; +import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException; import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer.TransactionFlushCallback; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor; import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry; @@ -235,12 +237,14 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle if (parallel) { // build stage processor multiStageCoprocessor = buildMultiStageCoprocessor(); - multiStageCoprocessor.start(); - if (isGTIDMode()) { // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 - erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), multiStageCoprocessor); + GTIDSet gtidSet = MysqlGTIDSet.parse(startPosition.getGtid()); + ((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet); + multiStageCoprocessor.start(); + erosaConnection.dump(gtidSet, multiStageCoprocessor); } else { + multiStageCoprocessor.start(); if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) { erosaConnection.dump(startPosition.getTimestamp(), multiStageCoprocessor); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ParserExceptionHandler.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ParserExceptionHandler.java index d0face01..3a77291d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ParserExceptionHandler.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ParserExceptionHandler.java @@ -5,5 +5,6 @@ package com.alibaba.otter.canal.parse.inbound; * @since 1.0.25 */ public interface ParserExceptionHandler { + void handle(Throwable e); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 941f5ebc..e36893a3 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -9,7 +9,6 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.filter.CanalEventFilter; import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; import com.alibaba.otter.canal.parse.CanalEventParser; -import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.AbstractEventParser; import com.alibaba.otter.canal.parse.inbound.BinlogParser; @@ -76,13 +75,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { * @return */ protected boolean processTableMeta(EntryPosition position) { - if (isGTIDMode()) { - if (binlogParser instanceof LogEventConvert) { - // 记录gtid - ((LogEventConvert) binlogParser).setGtidSet(MysqlGTIDSet.parse(position.getGtid())); - } - } - if (tableMetaTSDB != null) { if (position.getTimestamp() == null || position.getTimestamp() <= 0) { throw new CanalParseException("use gtid and TableMeta TSDB should be config timestamp > 0"); 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 eb3c0e31..4348dc29 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 @@ -4,6 +4,7 @@ import static com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetche import java.io.IOException; import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.nio.charset.Charset; import java.util.List; import java.util.concurrent.TimeUnit; @@ -179,6 +180,8 @@ public class MysqlConnection implements ErosaConnection { fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + // fix bug: #890 将gtid传输至context中，供decode使用 + context.setGtidSet(gtidSet); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); LogEvent event = null; @@ -252,10 +255,18 @@ public class MysqlConnection implements ErosaConnection { private void sendRegisterSlave() throws IOException { RegisterSlaveCommandPacket cmd = new RegisterSlaveCommandPacket(); - cmd.reportHost = authInfo.getAddress().getAddress().getHostAddress(); + SocketAddress socketAddress = connector.getChannel().getLocalSocketAddress(); + if (socketAddress == null || !(socketAddress instanceof InetSocketAddress)) { + return; + } + + InetSocketAddress address = (InetSocketAddress) socketAddress; + String host = address.getHostString(); + int port = address.getPort(); + cmd.reportHost = host; + cmd.reportPort = port; cmd.reportPasswd = authInfo.getPassword(); cmd.reportUser = authInfo.getUsername(); - cmd.reportPort = authInfo.getAddress().getPort(); // 暂时先用master节点的port cmd.serverId = this.slaveId; byte[] cmdBody = cmd.toBytes(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 722450dd..99bb94f5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -7,6 +7,7 @@ import java.util.concurrent.locks.LockSupport; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.utils.NamedThreadFactory; +import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer; @@ -63,6 +64,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private String destination; private volatile CanalParseException exception; private AtomicLong eventsPublishBlockingTime; + private GTIDSet gtidSet; public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, EventTransactionBuffer transactionBuffer, String destination){ @@ -81,8 +83,8 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement ringBufferSize, new BlockingWaitStrategy()); int tc = parserThreadCount > 0 ? parserThreadCount : 1; - this.parserExecutor = Executors.newFixedThreadPool(tc, - new NamedThreadFactory("MultiStageCoprocessor-Parser-" + destination)); + this.parserExecutor = Executors.newFixedThreadPool(tc, new NamedThreadFactory("MultiStageCoprocessor-Parser-" + + destination)); this.stageExecutor = Executors.newFixedThreadPool(2, new NamedThreadFactory("MultiStageCoprocessor-other-" + destination)); @@ -227,6 +229,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement public SimpleParserStage(){ decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); context = new LogContext(); + if (gtidSet != null) { + context.setGtidSet(gtidSet); + } } public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) throws Exception { @@ -447,4 +452,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement this.eventsPublishBlockingTime = eventsPublishBlockingTime; } + public void setGtidSet(GTIDSet gtidSet) { + this.gtidSet = gtidSet; + } } 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 cb579961..55090f65 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 @@ -17,7 +17,6 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; -import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException; import com.alibaba.otter.canal.parse.inbound.BinlogParser; @@ -99,13 +98,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private boolean filterRows = false; private boolean useDruidDdlFilter = true; - // latest gtid - private GTIDSet gtidSet; - - public LogEventConvert(GTIDSet gtidSet){ - this.gtidSet = gtidSet; - } - public LogEventConvert(){ } @@ -172,10 +164,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private Entry parseGTIDLogEvent(GtidLogEvent logEvent) { LogHeader logHeader = logEvent.getHeader(); - String value = logEvent.getSid().toString() + ":" + logEvent.getGno(); Pair.Builder builder = Pair.newBuilder(); builder.setKey("gtid"); - builder.setValue(value); + builder.setValue(logEvent.getGtidStr()); if (logEvent.getLastCommitted() != null) { builder.setKey("lastCommitted"); @@ -974,8 +965,4 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setFilterRows(boolean filterRows) { this.filterRows = filterRows; } - - public void setGtidSet(GTIDSet gtidSet) { - this.gtidSet = gtidSet; - } } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogDumpPerformanceTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogDumpPerformanceTest.java index 8ca9b044..25c88764 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogDumpPerformanceTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogDumpPerformanceTest.java @@ -20,12 +20,12 @@ public class MysqlBinlogDumpPerformanceTest { public static void main(String args[]) { final MysqlEventParser controller = new MysqlEventParser(); - final EntryPosition startPosition = new EntryPosition("mysql-bin.001699", 120L, 100L); + final EntryPosition startPosition = new EntryPosition("mysql-bin.000007", 89796293L, 100L); controller.setConnectionCharset(Charset.forName("UTF-8")); controller.setSlaveId(3344L); controller.setDetectingEnable(false); controller.setFilterQueryDml(true); - controller.setMasterInfo(new AuthenticationInfo(new InetSocketAddress("127.0.0.1", 3328), "root", "hello")); + controller.setMasterInfo(new AuthenticationInfo(new InetSocketAddress("100.81.154.142", 3306), "canal", "canal")); controller.setMasterPosition(startPosition); controller.setEnableTsdb(false); controller.setDestination("example"); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java index f0aaf67a..ed528d86 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java @@ -226,7 +226,7 @@ public class SimpleDdlParserTest { Assert.assertEquals("retl", result.getSchemaName()); Assert.assertEquals("retl_mark", result.getTableName()); - //test index name contains 'on' -- version + // test index name contains 'on' -- version queryString = "create index schema_new_index_version_s_idx on q_contract_account (contract_id,main_contract_id)"; result = SimpleDdlParser.parse(queryString, "retl"); Assert.assertNotNull(result); diff --git a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java index fb74d5f8..6c6d08ff 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/memory/MemoryEventStoreWithBuffer.java @@ -435,7 +435,7 @@ public class MemoryEventStoreWithBuffer extends AbstractCanalStoreScavenge imple if (batchMode.isMemSize()) { ackMemSize.addAndGet(memsize); // 尝试清空buffer中的内存，将ack之前的内存全部释放掉 - for (long index = sequence + 1; index <= next; index++) { + for (long index = sequence + 1; index < next; index++) { entries[getIndex(index)] = null;// 设置为null } }