docs: private EntryPosition findAsPerTimestampInSpecificLogFile(MysqlConnection mysqlConnection,
                                                                final Long startTimestamp,
                                                                final EntryPosition endPosition,
                                                                final String searchBinlogFile,
                                                                final Boolean justForPositionTimestamp) {
  
          final LogPosition logPosition = new LogPosition();
          try {
              mysqlConnection.reconnect();
              // 开始遍历文件
              mysqlConnection.seek(searchBinlogFile, 4L, endPosition.getGtid(), new SinkFunction<LogEvent>() {
  
                  private LogPosition lastPosition;
  
                  public boolean sink(LogEvent event) {
                      EntryPosition entryPosition = null;
                      try {
                          CanalEntry.Entry entry = parseAndProfilingIfNecessary(event, true);
                          if (justForPositionTimestamp && logPosition.getPostion() == null && event.getWhen() > 0) {
                              // 初始位点
                              entryPosition = new EntryPosition(searchBinlogFile,
                                  event.getLogPos() - event.getEventLen(),
                                  event.getWhen() * 1000,
                                  event.getServerId());
                              entryPosition.setGtid(event.getHeader().getGtidSetStr());
                              logPosition.setPostion(entryPosition);
                          }
  
                          // 直接用event的位点来处理,解决一个binlog文件里没有任何事件导致死循环无法退出的问题
                          String logfilename = event.getHeader().getLogFileName();
                          // 记录的是binlog end offest,
                          // 因为与其对比的offest是show master status里的end offest
                          Long logfileoffset = event.getHeader().getLogPos();
                          Long logposTimestamp = event.getHeader().getWhen() * 1000;
                          Long serverId = event.getHeader().getServerId();
  
                          // 如果最小的一条记录都不满足条件，可直接退出
                          if (logposTimestamp >= startTimestamp) {
                              return false;
                          }
  
                          if (StringUtils.equals(endPosition.getJournalName(), logfilename)
                              && endPosition.getPosition() <= logfileoffset) {
                              return false;
                          }
  
                          if (entry == null) {
                              return true;
                          }
  
                          // 记录一下上一个事务结束的位置，即下一个事务的position
                          // position = current +
                          // data.length，代表该事务的下一条offest，避免多余的事务重复
                          if (CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) {
                              entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp, serverId);
                              if (logger.isDebugEnabled()) {
                                  logger.debug("set {} to be pending start position before finding another proper one...",
                                      entryPosition);
                              }
                              logPosition.setPostion(entryPosition);
                              entryPosition.setGtid(entry.getHeader().getGtid());
                          } else if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType())) {
                              // 当前事务开始位点
                              entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp, serverId);
                              if (logger.isDebugEnabled()) {
                                  logger.debug("set {} to be pending start position before finding another proper one...",
                                      entryPosition);
                              }
                              entryPosition.setGtid(entry.getHeader().getGtid());
                              logPosition.setPostion(entryPosition);
                          }
  
                          lastPosition = buildLastPosition(entry);
                      } catch (Throwable e) {
                          processSinkError(e, lastPosition, searchBinlogFile, 4L);
                      }
  
                      return running;
                  }
              });
  
          } catch (IOException e) {
              logger.error("ERROR ## findAsPerTimestampInSpecificLogFile has an error", e);
          }
  
          if (logPosition.getPostion() != null) {
              return logPosition.getPostion();
          } else {
              return null;
          }
      }