docs: public EtlEventData await(Long pipelineId) throws InterruptedException {
          Assert.notNull(pipelineId);
          PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
          permitMonitor.waitForPermit();// 阻塞等待授权
  
          LoadStageListener loadStageListener = ArbitrateFactory.getInstance(pipelineId, LoadStageListener.class);
          Long processId = loadStageListener.waitForProcess(); // 符合条件的processId
  
          // DistributedLock lock = getLock(pipelineId);
          try {
              // 使用锁的理由：
              // 1. 针对双向同步时，其中一个方向出现了异常，需要发起另一端的关闭，此时对方正好在执行某个process的load
              // 2. 单向同步时，如果出现node节点异常，此时正常的节点正在执行某个process的load
              // 为避免因load无法中端引起的数据重复录入，所以针对load阶段添加分布式锁。在有process
              // load过程中不允许进行pipeline关闭操作
              // lock.lock();
  
              ChannelStatus status = permitMonitor.getChannelPermit();
              if (status.isStart()) {// 即时查询一下当前的状态，状态随时可能会变
                  // 根据pipelineId+processId构造对应的path
                  String path = StagePathUtils.getTransformStage(pipelineId, processId);
                  try {
                      byte[] data = zookeeper.readData(path);
                      return JsonUtils.unmarshalFromByte(data, EtlEventData.class);// 反序列化并返回
                  } catch (ZkNoNodeException e) {
                      logger.error("pipeline[{}] processId[{}] is invalid , retry again", pipelineId, processId);
                      // try {
                      // lock.unlock();// 出现任何异常解除lock
                      // } catch (KeeperException e1) {
                      // // ignore
                      // }
                      return await(pipelineId);// /出现节点不存在，说明出现了error情况,递归调用重新获取一次
                  } catch (ZkException e) {
                      throw e;
                  }
              } else {
                  logger.warn("pipelineId[{}] load ignore processId[{}] by status[{}]", new Object[] { pipelineId,
                          processId, status });
                  // try {
                  // lock.unlock();// 出现任何异常解除lock
                  // } catch (KeeperException e) {
                  // // ignore
                  // }
                  // 释放下processId，因为load是等待processId最小值完成Tranform才继续，如果这里不释放，会一直卡死等待
                  String path = StagePathUtils.getProcess(pipelineId, processId);
                  zookeeper.delete(path);
                  return await(pipelineId);// 出现rollback情况，递归调用重新获取一次，当前的processId可丢弃
              }
          } catch (InterruptedException e) {
              throw e;
          } catch (Exception e) {
              // try {
              // lock.unlock();// 出现任何异常解除lock
              // } catch (KeeperException e1) {
              // // ignore
              // }
  
              throw new ArbitrateException(e);
          }
      }