docs: public EtlEventData await(Long pipelineId) throws InterruptedException {
          Assert.notNull(pipelineId);
          PermitMonitor permitMonitor = ArbitrateFactory.getInstance(pipelineId, PermitMonitor.class);
          permitMonitor.waitForPermit();// 阻塞等待授权
  
          ExtractStageListener extractStageListener = ArbitrateFactory.getInstance(pipelineId, ExtractStageListener.class);
          Long processId = extractStageListener.waitForProcess(); // 符合条件的processId
  
          ChannelStatus status = permitMonitor.getChannelPermit();
          if (status.isStart()) {// 即时查询一下当前的状态，状态随时可能会变
              // 根据pipelineId+processId构造对应的path
              String path = StagePathUtils.getSelectStage(pipelineId, processId);
  
              try {
                  byte[] data = zookeeper.readData(path);
                  EtlEventData eventData = JsonUtils.unmarshalFromByte(data, EtlEventData.class);
  
                  Node node = LoadBalanceFactory.getNextTransformNode(pipelineId);// 获取下一个处理节点信息
                  if (node == null) {// 没有后端节点
                      // TerminEventData termin = new TerminEventData();
                      // termin.setPipelineId(pipelineId);
                      // termin.setType(TerminType.ROLLBACK);
                      // termin.setCode("no_node");
                      // termin.setDesc(MessageFormat.format("pipeline[{}] extract stage has no node!", pipelineId));
                      // terminEvent.single(termin);
                      throw new ArbitrateException("Extract_single", "no next node");
                  } else {
                      eventData.setNextNid(node.getId());
                      return eventData;// 只有这一条路返回
                  }
              } catch (ZkNoNodeException e) {
                  logger.error("pipeline[{}] processId[{}] is invalid , retry again", pipelineId, processId);
                  return await(pipelineId);// /出现节点不存在，说明出现了error情况,递归调用重新获取一次
              } catch (ZkException e) {
                  throw new ArbitrateException("Extract_await", e.getMessage(), e);
              }
          } else {
              logger.warn("pipelineId[{}] extract ignore processId[{}] by status[{}]", new Object[] { pipelineId,
                      processId, status });
                      
              // 释放下processId，因为load是等待processId最小值完成Tranform才继续，如果这里不释放，会一直卡死等待
              String path = StagePathUtils.getProcess(pipelineId, processId);
              zookeeper.delete(path);
              return await(pipelineId);// 递归调用
          }
  
      }