From b59984647093ef39bbd3c2801fd0b435a5ed7aa4 Mon Sep 17 00:00:00 2001 From: "Yishun.Chen" Date: Tue, 3 May 2016 16:29:23 +0800 Subject: [PATCH 1/3] =?UTF-8?q?IOException=E9=99=90=E5=88=B6retry=E6=AC=A1?= =?UTF-8?q?=E6=95=B0=EF=BC=8C=E8=B6=85=E8=BF=87=E6=AC=A1=E6=95=B0=E5=90=8E?= =?UTF-8?q?=E8=B7=B3=E8=BF=87binlog=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../parse/inbound/AbstractEventParser.java | 33 ++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) 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 b8c7f4b0..9cc6d3e3 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 @@ -87,12 +87,22 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle protected volatile Timer timer; protected TimerTask heartBeatTimerTask; protected Throwable exception = null; + + //update by yishun.chen + //特殊异常处理参数 + protected int specialExceptionCount = 0;//特殊异常计数 + protected int specialExceptionCountThreshold = 3;//特殊异常计数阀值 + protected boolean isFindEndPosition = false;//重连时查找数据库最新的位点 protected abstract BinlogParser buildParser(); protected abstract ErosaConnection buildErosaConnection(); protected abstract EntryPosition findStartPosition(ErosaConnection connection) throws IOException; + + //update by yishun.chen + //查找数据库最新的位点 + protected abstract EntryPosition findEndPosition(ErosaConnection connection) throws IOException; protected void preDump(ErosaConnection connection) { } @@ -159,7 +169,19 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle erosaConnection.connect();// 链接 // 4. 获取最后的位置信息 - final EntryPosition startPosition = findStartPosition(erosaConnection); + EntryPosition position = findStartPosition(erosaConnection); + //update by yishun.chen + //获取当前数据库最后一个位置进行消费 + //当解析binlog多次失败后跳过当前失败的位点重新到数据库获取新的位点 + if(isFindEndPosition){ + position = findEndPosition(erosaConnection); + specialExceptionCount = 0; + isFindEndPosition = false; + logger.error("special exception count>" + specialExceptionCountThreshold + ", find end position, maybe cause data loss!"); + sendAlarm(destination, "special exception count>" + specialExceptionCountThreshold + ", find end position, maybe cause data loss!"); + } + + final EntryPosition startPosition = position; if (startPosition == null) { throw new CanalParseException("can't find start position for " + destination); } @@ -219,6 +241,15 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle logger.error(String.format("dump address %s has an error, retrying. caused by ", runningInfo.getAddress().toString()), e); } catch (Throwable e) { + //update by yishun.chen + //dump数据出现IOException累计错误出现3次以上跳过当前位点 + if(e instanceof IOException){ + specialExceptionCount ++; + if(specialExceptionCount >= specialExceptionCountThreshold){ + isFindEndPosition = true; + } + } + exception = e; if (!running) { if (!(e instanceof java.nio.channels.ClosedByInterruptException || e.getCause() instanceof java.nio.channels.ClosedByInterruptException)) { From ecba855bf967fb9a374bea3d8872c1dacfa168f5 Mon Sep 17 00:00:00 2001 From: "Yishun.Chen" Date: Tue, 3 May 2016 16:30:01 +0800 Subject: [PATCH 2/3] =?UTF-8?q?IOException=E9=99=90=E5=88=B6retry=E6=AC=A1?= =?UTF-8?q?=E6=95=B0=EF=BC=8C=E8=B6=85=E8=BF=87=E6=AC=A1=E6=95=B0=E5=90=8E?= =?UTF-8?q?=E8=B7=B3=E8=BF=87binlog=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/parse/inbound/mysql/LocalBinlogEventParser.java | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java index 2317c0c2..61f4a53b 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.parse.inbound.mysql; +import java.io.IOException; + import org.apache.commons.lang.StringUtils; import com.alibaba.otter.canal.parse.CanalEventParser; @@ -52,6 +54,12 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements return connection; } + + @Override + protected EntryPosition findEndPosition(ErosaConnection connection) throws IOException { + // TODO 没有实现 + return findStartPosition(connection); + } @Override protected EntryPosition findStartPosition(ErosaConnection connection) { From 63b3c4ab543c5b54b359ad20c396b9e7544d9856 Mon Sep 17 00:00:00 2001 From: "Yishun.Chen" Date: Tue, 3 May 2016 16:30:38 +0800 Subject: [PATCH 3/3] =?UTF-8?q?IOException=E9=99=90=E5=88=B6retry=E6=AC=A1?= =?UTF-8?q?=E6=95=B0=EF=BC=8C=E8=B6=85=E8=BF=87=E6=AC=A1=E6=95=B0=E5=90=8E?= =?UTF-8?q?=E8=B7=B3=E8=BF=87binlog=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/parse/inbound/mysql/MysqlEventParser.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 d63f76fb..6deea962 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 @@ -322,6 +322,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } return startPosition; } + + protected EntryPosition findEndPosition(ErosaConnection connection) throws IOException { + MysqlConnection mysqlConnection = (MysqlConnection) connection; + EntryPosition endPosition = findEndPosition(mysqlConnection); + return endPosition; + } protected EntryPosition findStartPositionInternal(ErosaConnection connection) { MysqlConnection mysqlConnection = (MysqlConnection) connection;