docs: private void doTwoPhase(DbLoadContext context, List<List<EventData>> totalRows, boolean canBatch) {
          // 预处理下数据
          List<Future<Exception>> results = new ArrayList<Future<Exception>>();
          for (List<EventData> rows : totalRows) {
              if (CollectionUtils.isEmpty(rows)) {
                  continue; // 过滤空记录
              }
  
              results.add(executor.submit(new DbLoadWorker(context, rows, canBatch)));
          }
  
          boolean partFailed = false;
          for (int i = 0; i < results.size(); i++) {
              Future<Exception> result = results.get(i);
              Exception ex = null;
              try {
                  ex = result.get();
                  for (EventData data : totalRows.get(i)) {
                      interceptor.after(context, data);// 通知加载完成
                  }
              } catch (Exception e) {
                  ex = e;
              }
  
              if (ex != null) {
                  logger.warn("##load phase one failed!", ex);
                  partFailed = true;
              }
          }
  
          if (true == partFailed) {
              // if (CollectionUtils.isEmpty(context.getFailedDatas())) {
              // logger.error("##load phase one failed but failedDatas is empty!");
              // return;
              // }
  
              // 尝试的内容换成phase one跑的所有数据，避免因failed datas计算错误而导致丢数据
              List<EventData> retryEventDatas = new ArrayList<EventData>();
              for (List<EventData> rows : totalRows) {
                  retryEventDatas.addAll(rows);
              }
  
              context.getFailedDatas().clear(); // 清理failed data数据
  
              // 可能为null，manager老版本数据序列化传输时，因为数据库中没有skipLoadException变量配置
              Boolean skipLoadException = context.getPipeline().getParameters().getSkipLoadException();
              if (skipLoadException != null && skipLoadException) {// 如果设置为允许跳过单条异常，则一条条执行数据load，准确过滤掉出错的记录，并进行日志记录
                  for (EventData retryEventData : retryEventDatas) {
                      DbLoadWorker worker = new DbLoadWorker(context, Arrays.asList(retryEventData), false);// 强制设置batch为false
                      try {
                          Exception ex = worker.call();
                          if (ex != null) {
                              // do skip
                              logger.warn("skip exception for data : {} , caused by {}",
                                  retryEventData,
                                  ExceptionUtils.getFullStackTrace(ex));
                          }
                      } catch (Exception ex) {
                          // do skip
                          logger.warn("skip exception for data : {} , caused by {}",
                              retryEventData,
                              ExceptionUtils.getFullStackTrace(ex));
                      }
                  }
              } else {
                  // 直接一批进行处理，减少线程调度
                  DbLoadWorker worker = new DbLoadWorker(context, retryEventDatas, false);// 强制设置batch为false
                  try {
                      Exception ex = worker.call();
                      if (ex != null) {
                          throw ex; // 自己抛自己接
                      }
                  } catch (Exception ex) {
                      logger.error("##load phase two failed!", ex);
                      throw new LoadException(ex);
                  }
              }
  
              // 清理failed data数据
              for (EventData data : retryEventDatas) {
                  interceptor.after(context, data);// 通知加载完成
              }
          }
  
      }