将对应canal送出来的Entry对象解析为otter使用的内部对象

<pre>
需要处理数据过滤：
1. Transaction Begin/End过滤
2. retl.retl_client/retl.retl_mark 回环标记处理以及后续的回环数据过滤
3. retl.xdual canal心跳表数据过滤
</pre>
172012
public List<EventData> parse(Long pipelineId, List<Entry> datas) throws SelectException {
        List<EventData> eventDatas = new ArrayList<EventData>();
        Pipeline pipeline = configClientService.findPipeline(pipelineId);
        List<Entry> transactionDataBuffer = new ArrayList<Entry>();
        // hz为主站点，us->hz的数据，需要回环同步会us。并且需要开启回环补救算法
        PipelineParameter pipelineParameter = pipeline.getParameters();
        boolean enableLoopbackRemedy = pipelineParameter.isEnableRemedy() && pipelineParameter.isHome()
                                       && pipelineParameter.getRemedyAlgorithm().isLoopback();
        boolean isLoopback = false;
        boolean needLoopback = false; // 判断是否属于需要loopback处理的类型，只处理正常otter同步产生的回环数据，因为会有业务方手工屏蔽同步的接口，避免回环

        long now = new Date().getTime();
        try {
            for (Entry entry : datas) {
                switch (entry.getEntryType()) {
                    case TRANSACTIONBEGIN:
                        isLoopback = false;
                        break;
                    case ROWDATA:
                        String tableName = entry.getHeader().getTableName();
                        // 判断是否是回环表retl_mark
                        boolean isMarkTable = tableName.equalsIgnoreCase(pipeline.getParameters().getSystemMarkTable());
                        if (isMarkTable) {
                            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                            if (!rowChange.getIsDdl()) {
                                int loopback = 0;
                                if (rowChange.getRowDatasCount() > 0) {
                                    loopback = checkLoopback(pipeline, rowChange.getRowDatas(0));
                                }
                                if (loopback == 2) {
                                    needLoopback |= true; // 只处理正常同步产生的回环数据
                                }

                                isLoopback |= loopback > 0;
                            }
                        }

                        // 检查下otter3.0的回环表，对应的schmea会比较随意，所以不做比较
                        boolean isCompatibleLoopback = tableName.equalsIgnoreCase(compatibleMarkTable);

                        if (isCompatibleLoopback) {
                            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                            if (!rowChange.getIsDdl()) {
                                int loopback = 0;
                                if (rowChange.getRowDatasCount() > 0) {
                                    loopback = checkCompatibleLoopback(pipeline, rowChange.getRowDatas(0));
                                }
                                if (loopback == 2) {
                                    needLoopback |= true; // 只处理正常同步产生的回环数据
                                }
                                isLoopback |= loopback > 0;
                            }
                        }

                        if ((!isLoopback || (enableLoopbackRemedy && needLoopback)) && !isMarkTable
                            && !isCompatibleLoopback) {
                            transactionDataBuffer.add(entry);
                        }
                        break;
                    case TRANSACTIONEND:
                        if (!isLoopback || (enableLoopbackRemedy && needLoopback)) {
                            // 添加数据解析
                            for (Entry bufferEntry : transactionDataBuffer) {
                                List<EventData> parseDatas = internParse(pipeline, bufferEntry);
                                if (CollectionUtils.isEmpty(parseDatas)) {// 可能为空，针对ddl返回时就为null
                                    continue;
                                }

                                // 初步计算一下事件大小
                                long totalSize = bufferEntry.getHeader().getEventLength();
                                long eachSize = totalSize / parseDatas.size();
                                for (EventData eventData : parseDatas) {
                                    if (eventData == null) {
                                        continue;
                                    }

                                    eventData.setSize(eachSize);// 记录一下大小
                                    if (needLoopback) {// 针对需要回环同步的
                                        // 如果延迟超过指定的阀值，则设置为需要反查db
                                        if (now - eventData.getExecuteTime() > 1000 * pipeline.getParameters()
                                            .getRemedyDelayThresoldForMedia()) {
                                            eventData.setSyncConsistency(SyncConsistency.MEDIA);
                                        } else {
                                            eventData.setSyncConsistency(SyncConsistency.BASE);
                                        }
                                        eventData.setRemedy(true);
                                    }
                                    eventDatas.add(eventData);
                                }
                            }
                        }

                        isLoopback = false;
                        needLoopback = false;
                        transactionDataBuffer.clear();
                        break;
                    default:
                        break;
                }
            }

            // 添加最后一次的数据，可能没有TRANSACTIONEND
            if (!isLoopback || (enableLoopbackRemedy && needLoopback)) {
                // 添加数据解析
                for (Entry bufferEntry : transactionDataBuffer) {
                    List<EventData> parseDatas = internParse(pipeline, bufferEntry);
                    if (CollectionUtils.isEmpty(parseDatas)) {// 可能为空，针对ddl返回时就为null
                        continue;
                    }

                    // 初步计算一下事件大小
                    long totalSize = bufferEntry.getHeader().getEventLength();
                    long eachSize = totalSize / parseDatas.size();
                    for (EventData eventData : parseDatas) {
                        if (eventData == null) {
                            continue;
                        }

                        eventData.setSize(eachSize);// 记录一下大小
                        if (needLoopback) {// 针对需要回环同步的
                            // 如果延迟超过指定的阀值，则设置为需要反查db
                            if (now - eventData.getExecuteTime() > 1000 * pipeline.getParameters()
                                .getRemedyDelayThresoldForMedia()) {
                                eventData.setSyncConsistency(SyncConsistency.MEDIA);
                            } else {
                                eventData.setSyncConsistency(SyncConsistency.BASE);
                            }
                        }
                        eventDatas.add(eventData);
                    }
                }
            }
        } catch (Exception e) {
            throw new SelectException(e);
        }

        return eventDatas;
    }
994200
public RayTraceResult trace()
	{
		RayTraceResult mop = null, firstHit = null;
		double tX, tY, tZ, min;
		int count = 0;
		boolean ret = false;

		int currentX = blockSrc.getX();
		int currentY = blockSrc.getY();
		int currentZ = blockSrc.getZ();

		while (!ret && count++ <= MAX_BLOCKS)
		{
			tX = ray.intersectX(currentX + (ray.direction.x > 0 ? 1 : 0));
			tY = ray.intersectY(currentY + (ray.direction.y > 0 ? 1 : 0));
			tZ = ray.intersectZ(currentZ + (ray.direction.z > 0 ? 1 : 0));

			min = getMin(tX, tY, tZ);

			// do not trace first block
			if (count != 1 || !hasOption(Options.IGNORE_FIRST_BLOCK))
				mop = rayTraceBlock(new BlockPos(currentX, currentY, currentZ), ray.getPointAt(min));
			if (firstHit == null)
				firstHit = mop;
			if (hasOption(Options.LOG_BLOCK_PASSED))
				blockPassed.put(new BlockPos(currentX, currentY, currentZ), mop);

			if (dest != null && currentX == blockDest.getX() && currentY == blockDest.getY() && currentZ == blockDest.getZ())
				ret = true;

			if (!ret)
			{
				if (min == tX)
					currentX += step.x;
				if (min == tY)
					currentY += step.y;
				if (min == tZ)
					currentZ += step.z;
			}

			if (dest != null && dest.equals(ray.getPointAt(min)))
				ret = true;

		}

		if (firstHit == null && dest != null)
			firstHit = new RayTraceResult(RayTraceResult.Type.MISS, dest.toVec3d(), null, new BlockPos(currentX, currentY, currentZ));

		firstHit = ChunkCollision.get().getRayTraceResult(	world,
															Pair.of(src, dest),
															firstHit,
															hasOption(Options.HIT_LIQUIDS),
															hasOption(Options.CHECK_COLLISION),
															true);

		if (!ret)
			MalisisCore.message("Trace fail : " + MAX_BLOCKS + " blocks passed (" + currentX + "," + currentY + "," + currentZ + ")");
		return firstHit;
	}
265576
function logentriesRawStream(pathname, query={}) {
    query = Object.assign({}, query);

    // If query format is not defined, we set it by default to JSON.
    if (!query.format) {
        query.format = 'json';
    }

    // Text format is default in Logentries, in this case we remove the parameter.
    if (query.format === 'text') {
        delete query.format;
    }

    // Put default start and/or end times, if not given.
    if (!query.start && !query.end) {
        query.end   = Date.now();
        query.start = query.end - 15*60*1000; // 15 minutes
    } else if (!query.start) {
        query.start = query.end - 15*60*1000;
    } else if (!query.end) {
        query.end = query.start + 15*60*1000;
    }

    // Return raw stream using the download API.
    return request.get({
        url     : url.format({
            protocol : 'https:',
            host     : 'pull.logentries.com',
            pathname : pathname,
            query    : query
        }),
        gzip    : true,
        timeout : 30000 // 30 seconds
    });
}
1322472
private UmsLine createLine(ReportEntry1 entry, BigDecimal currSaldo) {
        UmsLine line = new UmsLine();
        line.isSepa = true;
        line.isCamt = true;
        line.other = new Konto();

        List<EntryTransaction1> txList = entry.getTxDtls();
        if (txList.size() == 0)
            return null;

        // Checken, ob es Soll- oder Habenbuchung ist
        boolean haben = entry.getCdtDbtInd() != null && entry.getCdtDbtInd() == CreditDebitCode.CRDT;

        // ditto
        EntryTransaction1 tx = txList.get(0);

        ////////////////////////////////////////////////////////////////////////
        // Buchungs-ID
        TransactionReferences1 ref = tx.getRefs();
        if (ref != null) {
            line.id = trim(ref.getPrtry() != null ? ref.getPrtry().getRef() : null);
            line.endToEndId = trim(ref.getEndToEndId());
            line.mandateId = trim(ref.getMndtId());
        }
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Gegenkonto: IBAN + Name
        TransactionParty1 other = tx.getRltdPties();
        if (other != null) {
            CashAccount7 acc = haben ? other.getDbtrAcct() : other.getCdtrAcct();
            AccountIdentification3Choice id = acc != null ? acc.getId() : null;
            line.other.iban = trim(id != null ? id.getIBAN() : null);

            PartyIdentification8 name = haben ? other.getDbtr() : other.getCdtr();
            line.other.name = trim(name != null ? name.getNm() : null);

            // Abweichender Name, falls vorhanden
            name = haben ? other.getUltmtDbtr() : other.getUltmtCdtr();
            line.other.name2 = trim(name != null ? name.getNm() : null);
        }
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Gegenkonto: BIC
        TransactionAgents1 banks = tx.getRltdAgts();
        if (banks != null) {
            BranchAndFinancialInstitutionIdentification3 bank = haben ? banks.getDbtrAgt() : banks.getCdtrAgt();
            FinancialInstitutionIdentification5Choice bic = bank != null ? bank.getFinInstnId() : null;
            line.other.bic = trim(bank != null ? bic.getBIC() : null);
        }
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Verwendungszweck
        List<String> usages = tx.getRmtInf() != null ? tx.getRmtInf().getUstrd() : null;
        if (usages != null && usages.size() > 0)
            line.usage.addAll(trim(usages));
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Betrag
        CurrencyAndAmount amt = entry.getAmt();
        BigDecimal bd = amt.getValue() != null ? amt.getValue() : BigDecimal.ZERO;
        line.value = new Value(this.checkDebit(bd, entry.getCdtDbtInd()));
        line.value.setCurr(amt.getCcy());
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Storno-Kennzeichen
        // Laut Spezifikation kehrt sich bei Stornobuchungen im Gegensatz zu MT940
        // nicht das Vorzeichen um. Der Betrag bleibt also gleich
        line.isStorno = entry.isRvslInd() != null && entry.isRvslInd().booleanValue();
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Buchungs- und Valuta-Datum
        DateAndDateTimeChoice bdate = entry.getBookgDt();
        line.bdate = bdate != null ? SepaUtil.toDate(bdate.getDt()) : null;

        DateAndDateTimeChoice vdate = entry.getValDt();
        line.valuta = vdate != null ? SepaUtil.toDate(vdate.getDt()) : null;

        // Wenn einer von beiden Werten fehlt, uebernehmen wir dort den jeweils anderen
        if (line.bdate == null) line.bdate = line.valuta;
        if (line.valuta == null) line.valuta = line.bdate;
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Saldo
        line.saldo = new Saldo();
        line.saldo.value = new Value(currSaldo.add(line.value.getBigDecimalValue()));
        line.saldo.value.setCurr(line.value.getCurr());
        line.saldo.timestamp = line.bdate;
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Art und Kundenreferenz
        line.text = trim(entry.getAddtlNtryInf());
        line.customerref = trim(entry.getAcctSvcrRef());
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Primanota, GV-Code und GV-Code-Ergaenzung
        // Ich weiss nicht, ob das bei allen Banken so codiert ist.
        // Bei der Sparkasse ist es jedenfalls so.
        BankTransactionCodeStructure1 b = tx.getBkTxCd();
        String code = (b != null && b.getPrtry() != null) ? b.getPrtry().getCd() : null;
        if (code != null && code.contains("+")) {
            String[] parts = code.split("\\+");
            if (parts.length == 4) {
                line.gvcode = parts[1];
                line.primanota = parts[2];
                line.addkey = parts[3];
            }
        }
        //
        ////////////////////////////////////////////////////////////////////////

        ////////////////////////////////////////////////////////////////////////
        // Purpose-Code
        Purpose1Choice purp = tx.getPurp();
        line.purposecode = trim(purp != null ? purp.getCd() : null);
        //
        ////////////////////////////////////////////////////////////////////////

        return line;
    }
1054056
public String getTransaction()
   {
      if (transaction == null)
      {
         TraceEvent transactionEvent =
            TraceEventHelper.getType(events,
                                     TraceEvent.ENLIST_CONNECTION_LISTENER,
                                     TraceEvent.ENLIST_CONNECTION_LISTENER_FAILED,
                                     TraceEvent.ENLIST_INTERLEAVING_CONNECTION_LISTENER,
                                     TraceEvent.ENLIST_INTERLEAVING_CONNECTION_LISTENER_FAILED);
      
         if (transactionEvent != null)
         {
            transaction = transactionEvent.getPayload1();
         }
         else
         {
            transaction = "";
         }
      }

      if (!"".equals(transaction))
         return transaction;

      return null;
   }
963844
@PostMapping("/etl/{type}/{key}/{task}")
    public EtlResult etl(@PathVariable String type, @PathVariable String key, @PathVariable String task,
                         @RequestParam(name = "params", required = false) String params) {
        OuterAdapter adapter = loader.getExtension(type, key);
        String destination = adapter.getDestination(task);
        String lockKey = destination == null ? task : destination;

        boolean locked = etlLock.tryLock(ETL_LOCK_ZK_NODE + type + "-" + lockKey);
        if (!locked) {
            EtlResult result = new EtlResult();
            result.setSucceeded(false);
            result.setErrorMessage(task + " 有其他进程正在导入中, 请稍后再试");
            return result;
        }
        try {

            boolean oriSwitchStatus;
            if (destination != null) {
                oriSwitchStatus = syncSwitch.status(destination);
                if (oriSwitchStatus) {
                    syncSwitch.off(destination);
                }
            } else {
                // task可能为destination，直接锁task
                oriSwitchStatus = syncSwitch.status(task);
                if (oriSwitchStatus) {
                    syncSwitch.off(task);
                }
            }
            try {
                List<String> paramArray = null;
                if (params != null) {
                    paramArray = Arrays.asList(params.trim().split(";"));
                }
                return adapter.etl(task, paramArray);
            } finally {
                if (destination != null && oriSwitchStatus) {
                    syncSwitch.on(destination);
                } else if (destination == null && oriSwitchStatus) {
                    syncSwitch.on(task);
                }
            }
        } finally {
            etlLock.unlock(ETL_LOCK_ZK_NODE + type + "-" + lockKey);
        }
    }
606240
@Override
    public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());

        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        if (!canalInstance.getMetaManager().isStart()) {
            canalInstance.getMetaManager().start();
        }

        canalInstance.getMetaManager().subscribe(clientIdentity); // 执行一下meta订阅

        Position position = canalInstance.getMetaManager().getCursor(clientIdentity);
        if (position == null) {
            position = canalInstance.getEventStore().getFirstPosition();// 获取一下store中的第一条
            if (position != null) {
                canalInstance.getMetaManager().updateCursor(clientIdentity, position); // 更新一下cursor
            }
            logger.info("subscribe successfully, {} with first position:{} ", clientIdentity, position);
        } else {
            logger.info("subscribe successfully, use last cursor position:{} ", clientIdentity, position);
        }

        // 通知下订阅关系变化
        canalInstance.subscribeChange(clientIdentity);
    }
513221
@Override
    public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }

        synchronized (canalInstance) {
            // 清除batch信息
            canalInstance.getMetaManager().clearAllBatchs(clientIdentity);
            // rollback eventStore中的状态信息
            canalInstance.getEventStore().rollback();
            logger.info("rollback successfully, clientId:{}", new Object[] { clientIdentity.getClientId() });
        }
    }
1202333
public synchronized void startLogSegment(RequestInfo reqInfo, long txid)
      throws IOException {
    assert fjm != null;
    checkJournalStorageFormatted();
    checkRequest(reqInfo);
    
    if (curSegment != null) {
      LOG.warn("Client is requesting a new log segment " + txid + 
          " though we are already writing " + curSegment + ". " +
          "Aborting the current segment in order to begin the new one.");
      // The writer may have lost a connection to us and is now
      // re-connecting after the connection came back.
      // We should abort our own old segment.
      abortCurSegment();
    }

    // Paranoid sanity check: we should never overwrite a finalized log file.
    // Additionally, if it's in-progress, it should have at most 1 transaction.
    // This can happen if the writer crashes exactly at the start of a segment.
    EditLogFile existing = fjm.getLogFile(txid);
    if (existing != null) {
      if (!existing.isInProgress()) {
        throw new IllegalStateException("Already have a finalized segment " +
            existing + " beginning at " + txid);
      }
      
      // If it's in-progress, it should only contain one transaction,
      // because the "startLogSegment" transaction is written alone at the
      // start of each segment. 
      existing.validateLog();
      if (existing.getLastTxId() != existing.getFirstTxId()) {
        throw new IllegalStateException("The log file " +
            existing + " seems to contain valid transactions");
      }
    }
    
    long curLastWriterEpoch = lastWriterEpoch.get();
    if (curLastWriterEpoch != reqInfo.getEpoch()) {
      LOG.info("Updating lastWriterEpoch from " + curLastWriterEpoch +
          " to " + reqInfo.getEpoch() + " for client " +
          Server.getRemoteIp());
      lastWriterEpoch.set(reqInfo.getEpoch());
      metrics.lastWriterEpoch.set(reqInfo.getEpoch());
    }

    // The fact that we are starting a segment at this txid indicates
    // that any previous recovery for this same segment was aborted.
    // Otherwise, no writer would have started writing. So, we can
    // remove the record of the older segment here.
    purgePaxosDecision(txid);
    
    curSegment = fjm.startLogSegment(txid);
    curSegmentTxId = txid;
    nextTxId = txid;
    // the layout version has only been written
    // plus OP_INVALID
    currentSegmentWrittenBytes = 5L;
    
    // inform the syncer service that we might have some work to do
    if (journalNode != null) {
      journalNode.addSyncTask(this, curSegmentTxId);
    }
  }
132584
@Override
    public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());

        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }
        synchronized (canalInstance) {
            // 清除batch信息
            PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
                batchId);
            if (positionRanges == null) { // 说明是重复的ack/rollback
                throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
                    clientIdentity.getClientId(),
                    batchId));
            }

            // lastRollbackPostions.put(clientIdentity,
            // positionRanges.getEnd());// 记录一下最后rollback的位置
            // TODO 后续rollback到指定的batchId位置
            canalInstance.getEventStore().rollback();// rollback
                                                     // eventStore中的状态信息
            logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
                clientIdentity.getClientId(),
                batchId,
                positionRanges);
        }
    }
101073
function getTransactionString(transaction) {
    var result = [];
    for (var key in transaction) {
        if (transaction[key] == null) {
            continue;
        }
        var value = bytes_1.hexlify(transaction[key]);
        if ({ gasLimit: true, gasPrice: true, nonce: true, value: true }[key]) {
            value = bytes_1.hexStripZeros(value);
        }
        result.push(key + '=' + value);
    }
    return result.join('&');
}
1141254
function parseBlockKatex(state, startLine, endLine) {
        var marker, len, params, nextLine, mem,
            haveEndMarker = false,
            pos = state.bMarks[startLine] + state.tShift[startLine],
            max = state.eMarks[startLine];
        var dollar = 0x24;

        if (pos + 1 > max) { return false; }

        marker = state.src.charCodeAt(pos);
        if (marker !== dollar) { return false; }

        // scan marker length
        mem = pos;
        pos = state.skipChars(pos, marker);
        len = pos - mem;

        if (len != 2)  { return false; }

        // search end of block
        nextLine = startLine;

        for (;;) {
            ++nextLine;
            if (nextLine >= endLine) {

                // unclosed block should be autoclosed by end of document.
                // also block seems to be autoclosed by end of parent
                break;
            }

            pos = mem = state.bMarks[nextLine] + state.tShift[nextLine];
            max = state.eMarks[nextLine];

            if (pos < max && state.tShift[nextLine] < state.blkIndent) {

                // non-empty line with negative indent should stop the list:
                // - ```
                //  test
                break;
            }

            if (state.src.charCodeAt(pos) !== dollar) { continue };

            if (state.tShift[nextLine] - state.blkIndent >= 4) {

                // closing fence should be indented less than 4 spaces
                continue;
            }

            pos = state.skipChars(pos, marker);

            // closing code fence must be at least as long as the opening one
            if (pos - mem < len) { continue; }

            // make sure tail has spaces only
            pos = state.skipSpaces(pos);

            if (pos < max) { continue; }

            haveEndMarker = true;

            // found!
            break;
        }

        // If a fence has heading spaces, they should be removed from its inner block
        len = state.tShift[startLine];

        state.line = nextLine + (haveEndMarker ? 1 : 0);
        
        var content = state.getLines(startLine + 1, nextLine, len, true)
                           .replace(/[ \n]+/g, ' ')
                           .trim();

        state.tokens.push({
            type: 'katex',
            params: params,
            content: content,
            lines: [startLine, state.line],
            level: state.level,
            block: true
        });

        return true;
    }
1019259
protected SITransaction getTransaction() {

        if (TRACE.isEntryEnabled()) {
            final String methodName = "getTransaction";
            SibTr.entry(this, TRACE, methodName);
            SibTr.exit(this, TRACE, methodName, null);
        }
        return null;

    }
765316
private void onConnectEnd(final int rstCode) {
        HMSAgentLog.d("connect end:" + rstCode);

        synchronized (CALLBACK_LOCK) {
            // 回调各个回调接口连接结束
            for (IClientConnectCallback callback : connCallbacks) {
                aSysnCallback(rstCode, callback);
            }
            connCallbacks.clear();

            // 恢复默认不显示
            allowResolveConnectError = false;
        }

        synchronized (STATIC_CALLBACK_LOCK) {
            // 回调各个回调接口连接结束
            for (IClientConnectCallback callback : staticCallbacks) {
                aSysnCallback(rstCode, callback);
            }
            staticCallbacks.clear();
        }
    }
440293
function TransactionSearch($startdate, $txnID = '', $email = '', $options) {
    if ($this->_mode == 'payflow') {
      $values['CUSTREF'] = $txnID;
      $values['TENDER'] = 'C';
      $values['TRXTYPE'] = 'I';
      $values['VERBOSITY'] = 'MEDIUM';
    } elseif ($this->_mode == 'nvp') {
      $values['STARTDATE'] = $startdate;
      $values['TRANSACTIONID'] = $txnID;
      $values['EMAIL'] = $email;
      if (is_array($options)) $values = array_merge($values, $options);
    }
    return $this->_request($values, 'TransactionSearch');
  }
1294640
@Override
    public Properties loadRemoteConfig() {
        Properties properties = null;
        try {
            // 加载远程canal配置
            ConfigItem configItem = getRemoteCanalConfig();
            if (configItem != null) {
                if (configItem.getModifiedTime() != currentConfigTimestamp) {
                    currentConfigTimestamp = configItem.getModifiedTime();
                    overrideLocalCanalConfig(configItem.getContent());
                    properties = new Properties();
                    properties.load(new ByteArrayInputStream(configItem.getContent().getBytes(StandardCharsets.UTF_8)));
                    scanIntervalInSecond = Integer
                        .valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5"));
                    logger.info("## Loaded remote canal config: canal.properties ");
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return properties;
    }
942534
public static TxInfo getTransactionInfo(Environment env, Bytes prow, Column pcol, long startTs) {
    // TODO ensure primary is visible

    IteratorSetting is = new IteratorSetting(10, RollbackCheckIterator.class);
    RollbackCheckIterator.setLocktime(is, startTs);

    Entry<Key, Value> entry = ColumnUtil.checkColumn(env, is, prow, pcol);

    TxInfo txInfo = new TxInfo();

    if (entry == null) {
      txInfo.status = TxStatus.UNKNOWN;
      return txInfo;
    }

    ColumnType colType = ColumnType.from(entry.getKey());
    long ts = entry.getKey().getTimestamp() & ColumnConstants.TIMESTAMP_MASK;

    switch (colType) {
      case LOCK: {
        if (ts == startTs) {
          txInfo.status = TxStatus.LOCKED;
          txInfo.lockValue = entry.getValue().get();
        } else {
          txInfo.status = TxStatus.UNKNOWN; // locked by another tx
        }
        break;
      }
      case DEL_LOCK: {
        DelLockValue dlv = new DelLockValue(entry.getValue().get());

        if (ts != startTs) {
          // expect this to always be false, must be a bug in the iterator
          throw new IllegalStateException(prow + " " + pcol + " (" + ts + " != " + startTs + ") ");
        }

        if (dlv.isRollback()) {
          txInfo.status = TxStatus.ROLLED_BACK;
        } else {
          txInfo.status = TxStatus.COMMITTED;
          txInfo.commitTs = dlv.getCommitTimestamp();
        }
        break;
      }
      case WRITE: {
        long timePtr = WriteValue.getTimestamp(entry.getValue().get());

        if (timePtr != startTs) {
          // expect this to always be false, must be a bug in the iterator
          throw new IllegalStateException(
              prow + " " + pcol + " (" + timePtr + " != " + startTs + ") ");
        }

        txInfo.status = TxStatus.COMMITTED;
        txInfo.commitTs = ts;
        break;
      }
      default:
        throw new IllegalStateException("unexpected col type returned " + colType);
    }

    return txInfo;
  }