From 911711248c4fdac09aeff8c3fe85406e5b472ce3 Mon Sep 17 00:00:00 2001 From: ruanjl Date: Sat, 9 Oct 2021 14:51:41 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Drds=E9=AB=98=E5=8F=AF?= =?UTF-8?q?=E7=94=A8=E4=B8=BB=E4=BB=8E=E5=88=87=E6=8D=A2=E5=92=8Coss?= =?UTF-8?q?=E7=9B=B8=E5=85=B3=E9=97=AE=E9=A2=98=20(#3480)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 修复rds oos 拉取的binlog 的消费逻辑以及本地消费到直连消费的逻辑衔接 * 去除无用代码 --- .../inbound/mysql/LocalBinLogConnection.java | 62 +++++++++++++++---- .../parse/inbound/mysql/MysqlEventParser.java | 4 ++ .../inbound/mysql/local/BinLogFileQueue.java | 12 ++-- .../mysql/rds/BinlogDownloadQueue.java | 6 +- .../mysql/rds/RdsLocalBinlogEventParser.java | 3 +- 5 files changed, 64 insertions(+), 23 deletions(-) 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; }