根据给定的时间戳，在指定的binlog中找到最接近于该时间戳(必须是小于时间戳)的一个事务起始位置。
针对最后一个binlog会给定endPosition，避免无尽的查询
770347
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;
        }
    }
191794
function end () {
          console.log('\n=== REG DUMP ===\n' + cpu.regdump().join('\n'))
          if (options.memDump) console.log('\n\n=== MEM DUMP ===\n' + cpu.memdump(0, 65535).join('\n'))
          process.exit() // workaround for stdin event listener hanging
        }
563135
public void addLog(final byte iOperation, final int iTxId, final int iClusterId, final long iClusterOffset,
      final byte iRecordType, final int iRecordVersion, final byte[] iRecordContent, int dataSegmentId) throws IOException {

    final int contentSize = iRecordContent != null ? iRecordContent.length : 0;
    final int size = OFFSET_RECORD_CONTENT + contentSize;

    lock.acquireExclusiveLock();
    try {

      int offset = file.allocateSpace(size);

      file.writeByte(offset, STATUS_COMMITTING);
      offset += OBinaryProtocol.SIZE_BYTE;

      file.writeByte(offset, iOperation);
      offset += OBinaryProtocol.SIZE_BYTE;

      file.writeInt(offset, iTxId);
      offset += OBinaryProtocol.SIZE_INT;

      file.writeShort(offset, (short) iClusterId);
      offset += OBinaryProtocol.SIZE_SHORT;

      file.writeLong(offset, iClusterOffset);
      offset += OBinaryProtocol.SIZE_LONG;

      file.writeByte(offset, iRecordType);
      offset += OBinaryProtocol.SIZE_BYTE;

      file.writeInt(offset, iRecordVersion);
      offset += OBinaryProtocol.SIZE_INT;

      file.writeInt(offset, dataSegmentId);
      offset += OBinaryProtocol.SIZE_INT;

      file.writeInt(offset, contentSize);
      offset += OBinaryProtocol.SIZE_INT;

      file.write(offset, iRecordContent);
      offset += contentSize;

      if (synchEnabled)
        file.synch();

    } finally {
      lock.releaseExclusiveLock();
    }
  }
1316627
private void onDelayModify(WatchEvent<?> event, Path currentPath) {
		Path eventPath = Paths.get(currentPath.toString(), event.context().toString());
		if(eventSet.contains(eventPath)) {
			//此事件已经被触发过，后续事件忽略，等待统一处理。
			return;
		}
		
		//事件第一次触发，此时标记事件，并启动处理线程延迟处理，处理结束后会删除标记
		eventSet.add(eventPath);
		startHandleModifyThread(event, currentPath);
	}
952142
public static int getEndPos(JCTree tree, EndPosTable endPosTable) {
        if (tree == null)
            return Position.NOPOS;

        if (endPosTable == null) {
            // fall back on limited info in the tree
            return endPos(tree);
        }

        int mapPos = endPosTable.getEndPos(tree);
        if (mapPos != Position.NOPOS)
            return mapPos;

        switch(tree.getTag()) {
            case BITOR_ASG: case BITXOR_ASG: case BITAND_ASG:
            case SL_ASG: case SR_ASG: case USR_ASG:
            case PLUS_ASG: case MINUS_ASG: case MUL_ASG:
            case DIV_ASG: case MOD_ASG:
                return getEndPos(((JCAssignOp) tree).rhs, endPosTable);
            case OR: case AND: case BITOR:
            case BITXOR: case BITAND: case EQ:
            case NE: case LT: case GT:
            case LE: case GE: case SL:
            case SR: case USR: case PLUS:
            case MINUS: case MUL: case DIV:
            case MOD:
                return getEndPos(((JCBinary) tree).rhs, endPosTable);
            case CASE:
                return getEndPos(((JCCase) tree).stats.last(), endPosTable);
            case CATCH:
                return getEndPos(((JCCatch) tree).body, endPosTable);
            case CONDEXPR:
                return getEndPos(((JCConditional) tree).falsepart, endPosTable);
            case FORLOOP:
                return getEndPos(((JCForLoop) tree).body, endPosTable);
            case FOREACHLOOP:
                return getEndPos(((JCEnhancedForLoop) tree).body, endPosTable);
            case IF: {
                JCIf node = (JCIf)tree;
                if (node.elsepart == null) {
                    return getEndPos(node.thenpart, endPosTable);
                } else {
                    return getEndPos(node.elsepart, endPosTable);
                }
            }
            case LABELLED:
                return getEndPos(((JCLabeledStatement) tree).body, endPosTable);
            case MODIFIERS:
                return getEndPos(((JCModifiers) tree).annotations.last(), endPosTable);
            case SYNCHRONIZED:
                return getEndPos(((JCSynchronized) tree).body, endPosTable);
            case TOPLEVEL:
                return getEndPos(((JCCompilationUnit) tree).defs.last(), endPosTable);
            case TRY: {
                JCTry node = (JCTry)tree;
                if (node.finalizer != null) {
                    return getEndPos(node.finalizer, endPosTable);
                } else if (!node.catchers.isEmpty()) {
                    return getEndPos(node.catchers.last(), endPosTable);
                } else {
                    return getEndPos(node.body, endPosTable);
                }
            }
            case WILDCARD:
                return getEndPos(((JCWildcard) tree).inner, endPosTable);
            case TYPECAST:
                return getEndPos(((JCTypeCast) tree).expr, endPosTable);
            case TYPETEST:
                return getEndPos(((JCInstanceOf) tree).clazz, endPosTable);
            case POS:
            case NEG:
            case NOT:
            case COMPL:
            case PREINC:
            case PREDEC:
                return getEndPos(((JCUnary) tree).arg, endPosTable);
            case WHILELOOP:
                return getEndPos(((JCWhileLoop) tree).body, endPosTable);
            case ANNOTATED_TYPE:
                return getEndPos(((JCAnnotatedType) tree).underlyingType, endPosTable);
            case ERRONEOUS: {
                JCErroneous node = (JCErroneous)tree;
                if (node.errs != null && node.errs.nonEmpty())
                    return getEndPos(node.errs.last(), endPosTable);
            }
        }
        return Position.NOPOS;
    }
891001
protected function end_log() {
        if (!$this->progress) {
            return;
        }
        echo get_string('done', 'tool_generator', round(microtime(true) - $this->starttime, 1));
        if (CLI_SCRIPT) {
            echo "\n";
        } else {
            echo html_writer::end_tag('li');
        }
    }
30656
function bintooctoutput(bin){
  var len = bin.length;
  var octoutput = String();
  for(var i=0; i<len; i+=3){
    if((i > 0) && !(i%24)){octoutput = " " + octoutput;}
    var value = 0;
    if(bin[len - 1 - i - 0]){value+=1;}
    if(bin[len - 1 - i - 1]){value+=2;}
    if(bin[len - 1 - i - 2]){value+=4;}
    switch(value){
      case 0: octoutput = "0" + octoutput; break;
      case 1: octoutput = "1" + octoutput; break;
      case 2: octoutput = "2" + octoutput; break;
      case 3: octoutput = "3" + octoutput; break;
      case 4: octoutput = "4" + octoutput; break;
      case 5: octoutput = "5" + octoutput; break;
      case 6: octoutput = "6" + octoutput; break;
      case 7: octoutput = "7" + octoutput; break;
      default: break;
    }
  }
  // todo: make faster
  return octoutput;
}
1352737
def _on_loop_end(self, variables):
        
        for callback in self.callbacks:
            if hasattr(callback, 'on_loop_end'):
                self.logs_[str(callback)].append(callback.on_loop_end(**variables))
996135
def log_end_message(log):
    
    if not log in MESSAGE_LOG.keys():
        log = -1

    res = colors.color_text(*MESSAGE_LOG[log][1])

    sys.stdout.write("\r[" + res + "] " + MESSAGE + "\n")
398473
private void perform(int operation, int position, long bs, int length) {
		checkBitsLength(length);
		position = adjPosition(position);
		checkMutable();
		performAdj(operation, position, bs, length);
	}
931740
protected void startEventParserInternal(CanalEventParser eventParser, boolean isGroup) {
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            CanalLogPositionManager logPositionManager = abstractEventParser.getLogPositionManager();
            if (!logPositionManager.isStart()) {
                logPositionManager.start();
            }
        }

        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            CanalHAController haController = mysqlEventParser.getHaController();

            if (haController instanceof HeartBeatHAController) {
                ((HeartBeatHAController) haController).setCanalHASwitchable(mysqlEventParser);
            }

            if (!haController.isStart()) {
                haController.start();
            }

        }
    }
1326580
public List<RawPacket> startBinlogDump(final int position, final String logfile) throws SQLException {
        try {
            return this.protocol.startBinlogDump(position, logfile);
        } catch (BinlogDumpException e) {
            throw SQLExceptionMapper.getSQLException("Could not dump binlog", e);
        }
    }
1186138
public static void closeStatement( Statement st, Logger logger ) {

		try {
			if( st != null )
				st.close();

		} catch( SQLException e ) {
			// Not important.
			Utils.logException( logger, e );
		}
	}
107161
@Override
    public boolean endWrite(OutputStream os) {
        boolean flag = true;
        try {
            this.workbook.write(os);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            CommonUtils.closeIOStream(null, os);
            //删除用于缓存的临时文件
            flag = ((SXSSFWorkbook) this.workbook).dispose();
        }
        return flag;
    }
585805
private boolean checkUnGetSlotAt(LogPosition startPosition, int batchSize) {
        if (batchMode.isItemSize()) {
            long current = getSequence.get();
            long maxAbleSequence = putSequence.get();
            long next = current;
            if (startPosition == null || !startPosition.getPostion().isIncluded()) { // 第一次订阅之后，需要包含一下start位置，防止丢失第一条记录
                next = next + 1;// 少一条数据
            }

            if (current < maxAbleSequence && next + batchSize - 1 <= maxAbleSequence) {
                return true;
            } else {
                return false;
            }
        } else {
            // 处理内存大小判断
            long currentSize = getMemSize.get();
            long maxAbleSize = putMemSize.get();

            if (maxAbleSize - currentSize >= batchSize * bufferMemUnit) {
                return true;
            } else {
                return false;
            }
        }
    }
1243202
def end(self, *args):
        

        if self._is_verbose:
            # verbose log has no end method
            return self

        if not args:
            self._indent -= 1
            return self
        self.writeln('end', *args)
        self._indent -= 1
        return self
315847
public File waitForNextFile(File pre) throws InterruptedException {
        try {
            lock.lockInterruptibly();
            if (binlogs.size() == 0) {
                nextCondition.await();// 等待新文件
            }

            if (exception != null) {
                throw exception;
            }
            if (pre == null) {// 第一次
                return binlogs.get(0);
            } else {
                int index = seek(pre);
                if (index < binlogs.size() - 1) {
                    return binlogs.get(index + 1);
                } else {
                    nextCondition.await();// 等待新文件
                    return waitForNextFile(pre);// 唤醒之后递归调用一下
                }
            }
        } finally {
            lock.unlock();
        }
    }