docs: private void syncStage(final Long processId) {
          // 1. 根据pipelineId + processId构造对应的path
          String path = null;
          try {
              path = StagePathUtils.getProcess(getPipelineId(), processId);
              // 2. 监听当前的process列表的变化
              IZkConnection connection = zookeeper.getConnection();
              // zkclient包装的是一个持久化的zk，分布式lock只需要一次性的watcher，需要调用原始的zk链接进行操作
              ZooKeeper orginZk = ((ZooKeeperx) connection).getZookeeper();
              List<String> currentStages = orginZk.getChildren(path, new AsyncWatcher() {
  
                  public void asyncProcess(WatchedEvent event) {
                      MDC.put(ArbitrateConstants.splitPipelineLogFileKey, String.valueOf(getPipelineId()));
                      if (isStop()) {
                          return;
                      }
  
                      if (event.getType() == EventType.NodeDeleted) {
                          processTermined(processId); // 触发下节点删除
                          return;
                      }
  
                      // 出现session expired/connection losscase下，会触发所有的watcher响应，同时老的watcher会继续保留，所以会导致出现多次watcher响应
                      boolean dataChanged = event.getType() == EventType.NodeDataChanged
                                            || event.getType() == EventType.NodeDeleted
                                            || event.getType() == EventType.NodeCreated
                                            || event.getType() == EventType.NodeChildrenChanged;
                      if (dataChanged) {
                          // boolean reply = initStage(processId);
                          // if (reply == false) {// 出现过load后就不需要再监听变化，剩下的就是节点的删除操作
                          syncStage(processId);
                          // }
                      }
                  }
              });
  
              Collections.sort(currentStages, new StageComparator());
              List<String> lastStages = this.currentStages.get(processId);
              if (lastStages == null || !lastStages.equals(currentStages)) {
                  initProcessStage(processId); // 存在差异，立马触发一下
              }
  
          } catch (NoNodeException e) {
              processTermined(processId); // 触发下节点删除
          } catch (KeeperException e) {
              syncStage(processId);
          } catch (InterruptedException e) {
              // ignore
          }
      }