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

fix bug #890

Browse Source
This commit is contained in:
winger
2018-09-03 15:28:15 +08:00
parent 21ef2e5dd4
commit 7de40091a3
8 changed files with 42 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<EVENT> 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);

7
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
View File

@@ -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");

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

@@ -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;

8
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
View File

@@ -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;
}
}

14
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
View File

@@ -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;
}
}