修复rds高可用主从切换和oss相关问题 (#3480)

* 修复rds oos 拉取的binlog 的消费逻辑以及本地消费到直连消费的逻辑衔接

* 去除无用代码

Author: ruanjl

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java

@@ -2,7 +2,9 @@
 
 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 @@
 
 /**
  * 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<Long> 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 void dump(String binlogfilename, Long binlogPosition, SinkFunction func)
                     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 void dump(String binlogfilename, Long binlogPosition, SinkFunction func)
         }
     }
 
+    /**
+     * 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<File> currentBinlogs = binlogs.currentBinlogs();
         File current = currentBinlogs.get(currentBinlogs.size() - 1);
@@ -158,9 +199,7 @@ public void dump(long timestampMills, SinkFunction func) throws IOException {
                 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 void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocess
                     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 void dump(long timestampMills, MultiStageCoprocessor coprocessor) throws
                 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 long getServerId() {
 
     public void setServerId(long serverId) {
         this.serverId = serverId;
+        rdsOssMasterSlaveInfo.add(serverId);
     }
 
     public void setParserListener(FileParserListener parserListener) {

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java

@@ -464,6 +464,10 @@ protected EntryPosition findStartPositionInternal(ErosaConnection connection) {
                     }
 
                     if (specificLogFilePosition == null) {
+                        if (isRdsOssMode()) {
+                            // 如果binlog位点不存在，并且属于timestamp不为空,可以返回null走到oss binlog处理
+                            return null;
+                        }
                         // position不存在，从文件头开始
                         entryPosition.setPosition(BINLOG_START_OFFEST);
                         return entryPosition;

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java

@@ -18,14 +18,15 @@
 
 /**
  * 维护binlog文件列表
- * 
+ *
  * @author jianghang 2012-7-7 下午03:48:05
  * @version 1.0.0
  */
 public class BinLogFileQueue {
 
     private String              baseName       = "mysql-bin.";
     private List<File>          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 void run() {
 
     /**
      * 根据前一个文件，获取符合条件的下一个binlog文件
-     * 
+     *
      * @param pre
      * @return
      */
@@ -141,7 +142,7 @@ public File getBefore(File file) {
 
     /**
      * 根据前一个文件，获取符合条件的下一个binlog文件
-     * 
+     *
      * @param pre
      * @return
      * @throws InterruptedException
@@ -219,9 +220,8 @@ private List<File> listBinlogFiles() {
         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) {

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java

@@ -123,7 +123,7 @@ private void filter(String hostInstanceId) {
     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 @@ private static void saveFile(File parentFile, String fileName, HttpResponse resp
                 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 @@ private static void saveFile(File parentFile, String fileName, HttpResponse resp
                 }
                 tais.close();
             } else {
-                File file = new File(parentFile, fileName + ".tmp");
+                File file = new File(parentFile, fileName);
                 if (file.exists()) {
                     file.delete();
                 }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java

@@ -21,7 +21,7 @@
 
 /**
  * 基于rds binlog备份文件的复制
- * 
+ *
  * @author agapple 2017年10月15日 下午1:27:36
  * @since 1.0.25
  */
@@ -117,6 +117,7 @@ protected ErosaConnection buildErosaConnection() {
             localBinLogConnection.setNeedWait(true);
             localBinLogConnection.setServerId(serverId);
             localBinLogConnection.setParserListener(this);
+            localBinLogConnection.setRdsOssMode(true);
         }
         return connection;
     }