feature(canal):support start with timestamp position(empty journal_file,set position=0),#2594

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

@@ -493,12 +493,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                                     && logPosition.getPostion().getServerId() != null
                                     && !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
                     if (case2) {
-                        long timestamp = logPosition.getPostion().getTimestamp();
-                        long newStartTimestamp = timestamp - fallbackIntervalInSeconds * 1000;
-                        logger.warn("prepare to find start position by last position {}:{}:{}", new Object[] { "", "",
-                                logPosition.getPostion().getTimestamp() });
-                        EntryPosition findPosition = findByStartTimeStamp(mysqlConnection, newStartTimestamp);
-                        // 重新置为一下
+                        EntryPosition findPosition = fallbackFindByStartTimestamp(logPosition, mysqlConnection);
                         dumpErrorCount = 0;
                         return findPosition;
                     }
@@ -508,6 +503,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                         // 如果binlog位点不存在，并且属于timestamp不为空,可以返回null走到oss binlog处理
                         return null;
                     }
+                } else if (StringUtils.isBlank(logPosition.getPostion().getJournalName())
+                        && logPosition.getPostion().getPosition() <= 0
+                        && logPosition.getPostion().getTimestamp() > 0) {
+                    return fallbackFindByStartTimestamp(logPosition,mysqlConnection);
                 }
                 // 其余情况
                 logger.warn("prepare to find start position just last position\n {}",
@@ -523,6 +522,21 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         }
     }
 
+    /**
+     * find position by timestamp with a fallback interval seconds.
+     *
+     * @param logPosition
+     * @param mysqlConnection
+     * @return
+     */
+    protected EntryPosition fallbackFindByStartTimestamp(LogPosition logPosition,MysqlConnection mysqlConnection){
+        long timestamp = logPosition.getPostion().getTimestamp();
+        long newStartTimestamp = timestamp - fallbackIntervalInSeconds * 1000;
+        logger.warn("prepare to find start position by last position {}:{}:{}", new Object[] { "", "",
+                logPosition.getPostion().getTimestamp() });
+        return findByStartTimeStamp(mysqlConnection, newStartTimestamp);
+    }
+
     // 根据想要的position，可能这个position对应的记录为rowdata，需要找到事务头，避免丢数据
     // 主要考虑一个事务执行时间可能会几秒种，如果仅仅按照timestamp相同，则可能会丢失事务的前半部分数据
     private Long findTransactionBeginPosition(ErosaConnection mysqlConnection, final EntryPosition entryPosition)
@@ -934,7 +948,4 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
         this.rdsOssMode = rdsOssMode;
     }
 
-    public void setDumpErrorCount(int dumpErrorCount) {
-        this.dumpErrorCount = dumpErrorCount;
-    }
 }

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

@@ -12,11 +12,11 @@ import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
 
 /**
  * aliyun rds的binlog parser支持
- *
+ * 
  * <pre>
  * 注意点：aliyun的binlog会有定期清理并备份到oss上, 这里实现了一份自动下载oss+rds binlog的机制
  * </pre>
- *
+ * 
  * @author chengjin.lyf on 2018/7/20 上午10:52
  * @since 1.0.25
  */
@@ -88,9 +88,6 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
                         @Override
                         public void run() {
                             rdsLocalBinlogEventParser.stop();
-                            // empty the dump error count,or will go into local binlog mode again,with error
-                            // position,never get out,fixed by bucketli
-                            RdsBinlogEventParserProxy.this.setDumpErrorCount(0);
                             RdsBinlogEventParserProxy.this.start();
                         }
                     });