diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java index 457b5d3b..74b70d91 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java @@ -2,7 +2,9 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.io.File; import java.io.IOException; +import java.util.HashSet; import java.util.List; +import java.util.Set; import org.apache.commons.lang.NotImplementedException; import org.apache.commons.lang.StringUtils; @@ -25,7 +27,7 @@ import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; /** * local bin log connection (not real connection) - * + * * @author yuanzu Date: 12-9-27 Time: 下午6:14 */ public class LocalBinLogConnection implements ErosaConnection { @@ -37,11 +39,28 @@ public class LocalBinLogConnection implements ErosaConnection { private int bufferSize = 16 * 1024; private boolean running = false; private long serverId; + + /** rdsOosMode binlog 的 serverId 是两个 */ + private boolean isRdsOssMode = false; + + /** rdsOosMode 主从信息 */ + private final Set rdsOssMasterSlaveInfo = new HashSet<>(4); + + private boolean firstUpdateRdsOssMasterSlave = true; + private FileParserListener parserListener; public LocalBinLogConnection(){ } + public boolean isRdsOssMode() { + return isRdsOssMode; + } + + public void setRdsOssMode(boolean rdsOssMode) { + isRdsOssMode = rdsOssMode; + } + public LocalBinLogConnection(String directory, boolean needWait){ this.needWait = needWait; this.directory = directory; @@ -94,9 +113,7 @@ public class LocalBinLogConnection implements ErosaConnection { if (event == null) { continue; } - if (serverId != 0 && event.getServerId() != serverId) { - throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); - } + checkServerId(event); if (!func.sink(event)) { needContinue = false; @@ -131,6 +148,30 @@ public class LocalBinLogConnection implements ErosaConnection { } } + /** + * 1. 非 rdsOos 模式下需要要校验 serverId 是否一致 防止解析其他实例的 binlog + * 2. rdsOos 高可用模式下解析 binlog 会有两个 serverId,分别对应着主从节点 binlog解析出来的 serverId + * 主从的关系可能会变, 但是 serverId一直都会是这两个 serverId + * + * @param event + */ + private void checkServerId(LogEvent event) { + if (serverId != 0 && event.getServerId() != serverId) { + if (isRdsOssMode()) { + // 第一次添加主从信息 + if (firstUpdateRdsOssMasterSlave) { + firstUpdateRdsOssMasterSlave = false; + rdsOssMasterSlaveInfo.add(event.getServerId()); + } else if (!rdsOssMasterSlaveInfo.contains(event.getServerId())) { + // 主从节点信息之外的节点信息 + throw new ServerIdNotMatchException("unexpected rds serverId " + serverId + " in binlog file !"); + } + } else { + throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); + } + } + } + public void dump(long timestampMills, SinkFunction func) throws IOException { List currentBinlogs = binlogs.currentBinlogs(); File current = currentBinlogs.get(currentBinlogs.size() - 1); @@ -158,9 +199,7 @@ public class LocalBinLogConnection implements ErosaConnection { while (fetcher.fetch()) { LogEvent event = decoder.decode(fetcher, context); if (event != null) { - if (serverId != 0 && event.getServerId() != serverId) { - throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); - } + checkServerId(event); if (event.getWhen() > timestampSeconds) { break; @@ -234,9 +273,7 @@ public class LocalBinLogConnection implements ErosaConnection { if (event == null) { continue; } - if (serverId != 0 && event.getServerId() != serverId) { - throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); - } + checkServerId(event); if (!coprocessor.publish(event)) { needContinue = false; @@ -304,9 +341,7 @@ public class LocalBinLogConnection implements ErosaConnection { while (fetcher.fetch()) { LogEvent event = decoder.decode(fetcher, context); if (event != null) { - if (serverId != 0 && event.getServerId() != serverId) { - throw new ServerIdNotMatchException("unexpected serverId " + serverId + " in binlog file !"); - } + checkServerId(event); if (event.getWhen() > timestampSeconds) { break; @@ -404,6 +439,7 @@ public class LocalBinLogConnection implements ErosaConnection { public void setServerId(long serverId) { this.serverId = serverId; + rdsOssMasterSlaveInfo.add(serverId); } public void setParserListener(FileParserListener parserListener) { 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 b6872b81..4e404e59 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 @@ -464,6 +464,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } if (specificLogFilePosition == null) { + if (isRdsOssMode()) { + // 如果binlog位点不存在，并且属于timestamp不为空,可以返回null走到oss binlog处理 + return null; + } // position不存在，从文件头开始 entryPosition.setPosition(BINLOG_START_OFFEST); return entryPosition; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java index ab048f30..2bfeefa8 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java @@ -18,7 +18,7 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException; /** * 维护binlog文件列表 - * + * * @author jianghang 2012-7-7 下午03:48:05 * @version 1.0.0 */ @@ -26,6 +26,7 @@ public class BinLogFileQueue { private String baseName = "mysql-bin."; private List binlogs = new ArrayList<>(); + private Pattern binLogPattern = Pattern.compile(baseName + "\\d+$"); private File directory; private ReentrantLock lock = new ReentrantLock(); private Condition nextCondition = lock.newCondition(); @@ -77,7 +78,7 @@ public class BinLogFileQueue { /** * 根据前一个文件，获取符合条件的下一个binlog文件 - * + * * @param pre * @return */ @@ -141,7 +142,7 @@ public class BinLogFileQueue { /** * 根据前一个文件，获取符合条件的下一个binlog文件 - * + * * @param pre * @return * @throws InterruptedException @@ -219,9 +220,8 @@ public class BinLogFileQueue { files.addAll(FileUtils.listFiles(directory, new IOFileFilter() { public boolean accept(File file) { - Pattern pattern = Pattern.compile("\\d+$"); - Matcher matcher = pattern.matcher(file.getName()); - return file.getName().startsWith(baseName) && matcher.find(); + Matcher matcher = binLogPattern.matcher(file.getName()); + return matcher.find(); } public boolean accept(File dir, String name) { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java index bd2e2648..5bb9bc67 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java @@ -123,7 +123,7 @@ public class BinlogDownloadQueue { public boolean isLastFile(String fileName) { String needCompareName = lastDownload; if (StringUtils.isNotEmpty(needCompareName) && StringUtils.endsWith(needCompareName, "tar")) { - needCompareName = needCompareName.substring(0, needCompareName.indexOf(".")); + needCompareName = needCompareName.substring(0, needCompareName.lastIndexOf(".")); } return (needCompareName == null || fileName.equalsIgnoreCase(needCompareName)) && binlogList.isEmpty(); } @@ -223,7 +223,7 @@ public class BinlogDownloadQueue { TarArchiveEntry tarArchiveEntry = null; while ((tarArchiveEntry = tais.getNextTarEntry()) != null) { String name = tarArchiveEntry.getName(); - File tarFile = new File(parentFile, name + ".tmp"); + File tarFile = new File(parentFile, name); logger.info("start to download file " + tarFile.getName()); if (tarFile.exists()) { tarFile.delete(); @@ -244,7 +244,7 @@ public class BinlogDownloadQueue { } tais.close(); } else { - File file = new File(parentFile, fileName + ".tmp"); + File file = new File(parentFile, fileName); if (file.exists()) { file.delete(); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java index 98674d47..97987062 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java @@ -21,7 +21,7 @@ import com.alibaba.otter.canal.protocol.position.LogPosition; /** * 基于rds binlog备份文件的复制 - * + * * @author agapple 2017年10月15日 下午1:27:36 * @since 1.0.25 */ @@ -117,6 +117,7 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements localBinLogConnection.setNeedWait(true); localBinLogConnection.setServerId(serverId); localBinLogConnection.setParserListener(this); + localBinLogConnection.setRdsOssMode(true); } return connection; }