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 6be3de39..ceedd8a4 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 @@ -120,7 +120,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle throw new CanalParseException("consume failed!"); } - LogPosition position = buildLastTranasctionPosition(transaction); + LogPosition position = buildLastTransactionPosition(transaction); if (position != null) { // 可能position为空 logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position); } @@ -336,20 +336,11 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle return profilingEnabled.get(); } - protected LogPosition buildLastTranasctionPosition(List entries) { // 初始化一下 + protected LogPosition buildLastTransactionPosition(List entries) { // 初始化一下 for (int i = entries.size() - 1; i > 0; i--) { CanalEntry.Entry entry = entries.get(i); if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {// 尽量记录一个事务做为position - LogPosition logPosition = new LogPosition(); - EntryPosition position = new EntryPosition(); - position.setJournalName(entry.getHeader().getLogfileName()); - position.setPosition(entry.getHeader().getLogfileOffset()); - position.setTimestamp(entry.getHeader().getExecuteTime()); - logPosition.setPostion(position); - - LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L); - logPosition.setIdentity(identity); - return logPosition; + return buildLastPosition(entry); } } 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 b779e03c..d9377005 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 @@ -700,22 +700,22 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } } - protected Entry parseAndProfilingIfNecessary(LogEvent bod) throws Exception { - long startTs = -1; - boolean enabled = getProfilingEnabled(); - if (enabled) { - startTs = System.currentTimeMillis(); - } - CanalEntry.Entry event = binlogParser.parse(bod); - if (enabled) { - this.parsingInterval = System.currentTimeMillis() - startTs; - } - - if (parsedEventCount.incrementAndGet() < 0) { - parsedEventCount.set(0); - } - return event; - } +// protected Entry parseAndProfilingIfNecessary(LogEvent bod) throws Exception { +// long startTs = -1; +// boolean enabled = getProfilingEnabled(); +// if (enabled) { +// startTs = System.currentTimeMillis(); +// } +// CanalEntry.Entry event = binlogParser.parse(bod); +// if (enabled) { +// this.parsingInterval = System.currentTimeMillis() - startTs; +// } +// +// if (parsedEventCount.incrementAndGet() < 0) { +// parsedEventCount.set(0); +// } +// return event; +// } public void setSupportBinlogFormats(String formatStrs) { String[] formats = StringUtils.split(formatStrs, ',');