docs: 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;
      }