mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
Compare commits
1 Commits
canal-1.1.
...
startup_ti
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3249d3c50d |
@@ -493,12 +493,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
|
|||||||
&& logPosition.getPostion().getServerId() != null
|
&& logPosition.getPostion().getServerId() != null
|
||||||
&& !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
|
&& !logPosition.getPostion().getServerId().equals(findServerId(mysqlConnection));
|
||||||
if (case2) {
|
if (case2) {
|
||||||
long timestamp = logPosition.getPostion().getTimestamp();
|
EntryPosition findPosition = fallbackFindByStartTimestamp(logPosition, mysqlConnection);
|
||||||
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);
|
|
||||||
// 重新置为一下
|
|
||||||
dumpErrorCount = 0;
|
dumpErrorCount = 0;
|
||||||
return findPosition;
|
return findPosition;
|
||||||
}
|
}
|
||||||
@@ -508,6 +503,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
|
|||||||
// 如果binlog位点不存在,并且属于timestamp不为空,可以返回null走到oss binlog处理
|
// 如果binlog位点不存在,并且属于timestamp不为空,可以返回null走到oss binlog处理
|
||||||
return null;
|
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 {}",
|
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,需要找到事务头,避免丢数据
|
// 根据想要的position,可能这个position对应的记录为rowdata,需要找到事务头,避免丢数据
|
||||||
// 主要考虑一个事务执行时间可能会几秒种,如果仅仅按照timestamp相同,则可能会丢失事务的前半部分数据
|
// 主要考虑一个事务执行时间可能会几秒种,如果仅仅按照timestamp相同,则可能会丢失事务的前半部分数据
|
||||||
private Long findTransactionBeginPosition(ErosaConnection mysqlConnection, final EntryPosition entryPosition)
|
private Long findTransactionBeginPosition(ErosaConnection mysqlConnection, final EntryPosition entryPosition)
|
||||||
|
|||||||
Reference in New Issue
Block a user