From 7de40091a34aaef868fa1244dc29a46602f625e9 Mon Sep 17 00:00:00 2001 From: winger Date: Mon, 3 Sep 2018 15:28:15 +0800 Subject: [PATCH 1/5] fix bug #890 --- .../taobao/tddl/dbsync/binlog/LogContext.java | 15 ++++--------- .../taobao/tddl/dbsync/binlog/LogDecoder.java | 21 ++++++++++++------- .../dbsync/binlog/event/GtidLogEvent.java | 7 +++++++ .../parse/inbound/AbstractEventParser.java | 10 ++++++--- .../mysql/AbstractMysqlEventParser.java | 7 ------- .../parse/inbound/mysql/MysqlConnection.java | 2 ++ .../mysql/MysqlMultiStageCoprocessor.java | 8 +++++++ .../inbound/mysql/dbsync/LogEventConvert.java | 14 +------------ 8 files changed, 42 insertions(+), 42 deletions(-) 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..11005197 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 @@ -66,18 +66,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..c4a99852 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 @@ -4,6 +4,7 @@ import java.io.IOException; import java.util.BitSet; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; +import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -47,22 +48,22 @@ import com.taobao.tddl.dbsync.binlog.event.mariadb.StartEncryptionLogEvent; /** * Implements a binary-log decoder. - * + * * 
  * LogDecoder decoder = new LogDecoder();
  * decoder.handle(...);
- * 
+ *
  * LogEvent event;
  * do
  * {
  *     event = decoder.decode(buffer, context);
- * 
+ *
  *     // process log event.
  * }
  * while (event != null);
  * // no more events in buffer.
  *
- * + * * @author Changyuan.lh * @version 1.0 */ @@ -89,7 +90,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 +142,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 +161,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 +375,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..b2897d58 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 @@ -82,4 +82,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/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..be2354dd 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 @@ -8,6 +8,8 @@ import java.util.TimerTask; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.math.RandomUtils; @@ -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/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 941f5ebc..4883d971 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 @@ -76,13 +76,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..256b1ecc 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 @@ -179,6 +179,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; 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..89c6e842 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){ @@ -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..17c55e8d 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 @@ -99,13 +99,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 +165,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 +966,4 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setFilterRows(boolean filterRows) { this.filterRows = filterRows; } - - public void setGtidSet(GTIDSet gtidSet) { - this.gtidSet = gtidSet; - } } From 47ac10ca185ef128ccdd055cb83fea5def7428c3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Tue, 4 Sep 2018 11:28:56 +0800 Subject: [PATCH 2/5] fixed issue #868 , NPE fixed --- .../otter/canal/store/memory/MemoryEventStoreWithBuffer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 } } From a9cccb3017709f335c57d74354eef72ac97e9df8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Tue, 4 Sep 2018 11:39:23 +0800 Subject: [PATCH 3/5] fixed format --- .../com/taobao/tddl/dbsync/binlog/LogContext.java | 2 -- .../com/taobao/tddl/dbsync/binlog/LogDecoder.java | 7 +++---- .../tddl/dbsync/binlog/event/GtidLogEvent.java | 8 +++----- .../taobao/tddl/dbsync/FetcherPerformanceTest.java | 4 +--- .../parse/exception/PositionNotFoundException.java | 10 +++++----- .../parse/exception/ServerIdNotMatchException.java | 12 ++++++------ .../canal/parse/inbound/AbstractEventParser.java | 4 ++-- .../canal/parse/inbound/ParserExceptionHandler.java | 1 + .../inbound/mysql/AbstractMysqlEventParser.java | 1 - .../inbound/mysql/MysqlMultiStageCoprocessor.java | 4 ++-- .../parse/inbound/mysql/dbsync/LogEventConvert.java | 1 - .../parse/inbound/mysql/SimpleDdlParserTest.java | 2 +- 12 files changed, 24 insertions(+), 32 deletions(-) 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 11005197..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; /** 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 c4a99852..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,11 +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 com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; 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; @@ -52,12 +51,12 @@ import com.taobao.tddl.dbsync.binlog.event.mariadb.StartEncryptionLogEvent; * 
  * LogDecoder decoder = new LogDecoder();
  * decoder.handle(...);
- *
+ * 
  * LogEvent event;
  * do
  * {
  *     event = decoder.decode(buffer, context);
- *
+ * 
  *     // process log event.
  * }
  * while (event != null);
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 b2897d58..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;
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/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 be2354dd..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
@@ -8,8 +8,6 @@ import java.util.TimerTask;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
-import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
-import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.commons.lang.math.RandomUtils;
@@ -21,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;
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 4883d971..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;
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 89c6e842..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
@@ -83,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));
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 17c55e8d..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;
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);

From f5bf1667a9bdb5559532ea822615628a6dd68bef Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= 
Date: Tue, 4 Sep 2018 12:35:20 +0800
Subject: [PATCH 4/5] fixed mq and client launcher

---
 client-launcher/src/main/bin/startup.sh                   | 2 +-
 client-launcher/src/main/bin/stop.sh                      | 4 ++--
 .../com/alibaba/otter/canal/deployer/CanalController.java | 2 +-
 .../com/alibaba/otter/canal/deployer/CanalLauncher.java   | 8 +++++---
 deployer/src/main/resources/kafka.yml                     | 4 +---
 5 files changed, 10 insertions(+), 10 deletions(-)

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/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

From 82f8a9fc980dd428eb45c8f66789564632723897 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= 
Date: Tue, 4 Sep 2018 12:55:16 +0800
Subject: [PATCH 5/5] fixed issue #483 , show slave hosts

---
 .../parse/driver/mysql/socket/BioSocketChannel.java | 10 ++++++++++
 .../driver/mysql/socket/NettySocketChannel.java     |  4 ++++
 .../parse/driver/mysql/socket/SocketChannel.java    |  2 ++
 .../canal/parse/inbound/mysql/MysqlConnection.java  | 13 +++++++++++--
 .../canal/parse/MysqlBinlogDumpPerformanceTest.java |  4 ++--
 5 files changed, 29 insertions(+), 4 deletions(-)

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/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
index 256b1ecc..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;
@@ -254,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/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");