通过pipeLineId得到一个以gmtCreate倒排序的第一条记录
972294
public DelayStat findRealtimeDelayStat(Long pipelineId) {
        Assert.assertNotNull(pipelineId);
        DelayStatDO delayStatDO = delayStatDao.findRealtimeDelayStat(pipelineId);
        DelayStat delayStat = new DelayStat();
        if (delayStatDO != null) {
            delayStat = delayStatDOToModel(delayStatDO);
        }
        return delayStat;
    }
581902
private void initStage(List<String> currentProcesses) {
        // 3. 循环处理每个process
        List<Long> processIds = new ArrayList<Long>();
        for (String process : currentProcesses) {
            processIds.add(StagePathUtils.getProcessId(process));
        }
        Collections.sort(processIds); // 排序一下
        List<Long> needSyncProcessIds = new ArrayList<Long>();

        synchronized (currentProcessIds) { // 需要同步处理
            for (Long processId : processIds) {
                if (!currentProcessIds.contains(processId)) {
                    needSyncProcessIds.add(processId);
                }
            }

            for (Long currentProcessId : currentProcessIds) {
                if (!processIds.contains(currentProcessId)) {
                    currentStages.remove(currentProcessId);// 如果process已删除,删除本地的stage信息
                }
            }

            if (logger.isDebugEnabled()) {
                logger.debug("pipeline[{}] old processIds{},current processIds{}", new Object[] { getPipelineId(),
                        currentProcessIds, processIds });
            }

            currentProcessIds = processIds; // 切换引用，需设置为volatile保证线程安全&可见性
        }

        for (Long syncProcessId : needSyncProcessIds) {
            syncStage(syncProcessId);// 开始同步单个的process
            // boolean reply = initProcessStage(processId);// 通知下节点变化
            // if (!currentProcessIds.contains(processId) && reply == false) {
            // syncStage(processId);// 开始同步单个的process
            // } else {
            // // 如果上一次循环中已经添加了监控，则忽略掉本次处理, 避免重复监听
            // }
        }

        if (processIds.size() > 0) {// 尝试触发一下第一个process进行load
            initProcessStage(processIds.get(0));
        }
        processChanged(currentProcessIds);// 通知变化
    }
859879
function getRoute(exportFuncs, cb, ctrlpath, curCtrlname) {
  ctrlpath = ctrlpath || [];

  // 如果当前设置了不是路由，则直接返回
  if (exportFuncs.__controller__ === false) {
    return;
  }

  let totalCtrlname = curCtrlname ? ctrlpath.concat([curCtrlname]) : ctrlpath;

  // 只允许3级路由层级
  if (ctrlpath.length > 3) {
    debug(`嵌套路由对象层级不能超过3级：${totalCtrlname.join('/')}`);
    return;
  }

  // 如果是一个方法就直接执行cb
  if (typeof exportFuncs === 'function') {
    cb(exportFuncs, totalCtrlname);
  } else {
    // 否则进行循环递归查询
    for (let ctrlname in exportFuncs) {
      if (!exportFuncs.hasOwnProperty(ctrlname)) {
        continue
      }

      getRoute(exportFuncs[ctrlname], cb, totalCtrlname, ctrlname);
    }
  }
}
1154970
public static String getProcessRoot(Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_PROCESS_ROOT, getChannelId(pipelineId),
                                    String.valueOf(pipelineId));
    }
162810
function BuildLog(){
	var my;
	var nextSegmentId = 0;
	var curVert = false;

	function push(type, data){
		my.list.push({
			type: type,
			data: data ? JSON.parse(JSON.stringify(data)) : void 0
		});
		return my;
	}

	my = {
		list: [],
		segmentId: function(){
			return nextSegmentId++;
		},
		checkIntersection: function(seg1, seg2){
			return push('check', { seg1: seg1, seg2: seg2 });
		},
		segmentChop: function(seg, end){
			push('div_seg', { seg: seg, pt: end });
			return push('chop', { seg: seg, pt: end });
		},
		statusRemove: function(seg){
			return push('pop_seg', { seg: seg });
		},
		segmentUpdate: function(seg){
			return push('seg_update', { seg: seg });
		},
		segmentNew: function(seg, primary){
			return push('new_seg', { seg: seg, primary: primary });
		},
		segmentRemove: function(seg){
			return push('rem_seg', { seg: seg });
		},
		tempStatus: function(seg, above, below){
			return push('temp_status', { seg: seg, above: above, below: below });
		},
		rewind: function(seg){
			return push('rewind', { seg: seg });
		},
		status: function(seg, above, below){
			return push('status', { seg: seg, above: above, below: below });
		},
		vert: function(x){
			if (x === curVert)
				return my;
			curVert = x;
			return push('vert', { x: x });
		},
		log: function(data){
			if (typeof data !== 'string')
				data = JSON.stringify(data, false, '  ');
			return push('log', { txt: data });
		},
		reset: function(){
			return push('reset');
		},
		selected: function(segs){
			return push('selected', { segs: segs });
		},
		chainStart: function(seg){
			return push('chain_start', { seg: seg });
		},
		chainRemoveHead: function(index, pt){
			return push('chain_rem_head', { index: index, pt: pt });
		},
		chainRemoveTail: function(index, pt){
			return push('chain_rem_tail', { index: index, pt: pt });
		},
		chainNew: function(pt1, pt2){
			return push('chain_new', { pt1: pt1, pt2: pt2 });
		},
		chainMatch: function(index){
			return push('chain_match', { index: index });
		},
		chainClose: function(index){
			return push('chain_close', { index: index });
		},
		chainAddHead: function(index, pt){
			return push('chain_add_head', { index: index, pt: pt });
		},
		chainAddTail: function(index, pt){
			return push('chain_add_tail', { index: index, pt: pt, });
		},
		chainConnect: function(index1, index2){
			return push('chain_con', { index1: index1, index2: index2 });
		},
		chainReverse: function(index){
			return push('chain_rev', { index: index });
		},
		chainJoin: function(index1, index2){
			return push('chain_join', { index1: index1, index2: index2 });
		},
		done: function(){
			return push('done');
		}
	};
	return my;
}
655114
public static String getMainStem(Long channelId, Long pipelineId) {
        return MessageFormat.format(ArbitrateConstants.NODE_PIPELINE_FORMAT,
            String.valueOf(channelId),
            String.valueOf(pipelineId))
               + "/" + ArbitrateConstants.NODE_MAINSTEM;
    }
522549
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);
        }
    }
847307
function generateSidebar(statusMap, nextPipelineId, rootElement) {
  var currentElement = null;

  if (nextPipelineId) {
    currentElement = $('<li>');
    // Value should match return of getTreePipelineElementId
    currentElement.attr('id', 'item-pipeline-' + nextPipelineId);
  } else {
    currentElement = rootElement;
    nextPipelineId = statusMap.rootPipelineId;
  }

  var parentInfoMap = statusMap.pipelines[nextPipelineId];
  currentElement.append(
      constructStageNode(nextPipelineId, parentInfoMap, true));

  if (statusMap.pipelines[nextPipelineId]) {
    var children = statusMap.pipelines[nextPipelineId].children;
    if (children.length > 0) {
      var treeElement = null;
      if (rootElement) {
        treeElement =
            $('<ul id="pipeline-tree" class="treeview-black treeview">');
      } else {
        treeElement = $('<ul>');
      }

      $.each(children, function(index, childPipelineId) {
        var childElement = generateSidebar(statusMap, childPipelineId);
        treeElement.append(childElement);
      });
      currentElement.append(treeElement);
    }
  }
  return currentElement;
}
749533
private PipelineDO modelToDo(Pipeline pipeline) {
        PipelineDO pipelineDO = new PipelineDO();

        try {
            pipelineDO.setId(pipeline.getId());
            pipelineDO.setName(pipeline.getName());
            pipelineDO.setParameters(pipeline.getParameters());
            pipelineDO.setDescription(pipeline.getDescription());
            pipelineDO.setChannelId(pipeline.getChannelId());
            pipelineDO.setGmtCreate(pipeline.getGmtCreate());
            pipelineDO.setGmtModified(pipeline.getGmtModified());

        } catch (Exception e) {
            logger.error("ERROR ## change the pipeline Model to Do has an exception");
            throw new ManagerException(e);
        }

        return pipelineDO;
    }
945576
function createMachineID(val_bb3, val_ff2, val_3b3) {
	// Machine IDs are binary KV objects with root key MessageObject and three hashes named BB3, FF2, and 3B3.
	// I don't feel like writing a proper BinaryKV serializer, so this will work fine.

	let buffer = ByteBuffer.allocate(155, ByteBuffer.LITTLE_ENDIAN);
	buffer.writeByte(0); // 1 byte, total 1
	buffer.writeCString("MessageObject"); // 14 bytes, total 15

	buffer.writeByte(1); // 1 byte, total 16
	buffer.writeCString("BB3"); // 4 bytes, total 20
	buffer.writeCString(sha1(val_bb3)); // 41 bytes, total 61

	buffer.writeByte(1); // 1 byte, total 62
	buffer.writeCString("FF2"); // 4 bytes, total 66
	buffer.writeCString(sha1(val_ff2)); // 41 bytes, total 107

	buffer.writeByte(1); // 1 byte, total 108
	buffer.writeCString("3B3"); // 4 bytes, total 112
	buffer.writeCString(sha1(val_3b3)); // 41 bytes, total 153

	buffer.writeByte(8); // 1 byte, total 154
	buffer.writeByte(8); // 1 byte, total 155

	return buffer.flip().toBuffer();

	function sha1(input) {
		let hash = Crypto.createHash('sha1');
		hash.update(input, 'utf8');
		return hash.digest('hex');
	}
}
612176
public static String getProcess(Long pipelineId, Long processId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_PROCESS_FORMAT, getChannelId(pipelineId),
                                    String.valueOf(pipelineId), getProcessNode(processId));
    }
677573
def log_pipeline(self, pl):
        
        from datetime import datetime
        from ambry.etl.pipeline import CastColumns

        self.build_fs.makedir('pipeline', allow_recreate=True)

        try:
            ccp = pl[CastColumns]
            caster_code = ccp.pretty_code
        except Exception as e:
            caster_code = str(e)

        templ = u("""
Pipeline     : {}
run time     : {}
phase        : {}
source name  : {}
source table : {}
dest table   : {}
========================================================
{}

Pipeline Headers
================
{}

Caster Code
===========
{}

""")
        try:
            v = templ.format(pl.name, str(datetime.now()), pl.phase, pl.source_name, pl.source_table,
                             pl.dest_table, unicode(pl), pl.headers_report(), caster_code)
        except UnicodeError as e:
            v = ''
            self.error('Faled to write pipeline log for pipeline {} '.format(pl.name))

        path = os.path.join('pipeline', pl.phase + '-' + pl.file_name + '.txt')

        self.build_fs.makedir(os.path.dirname(path), allow_recreate=True, recursive=True)
        # LazyFS should handled differently because of:
        # TypeError: lazy_fs.setcontents(..., encoding='utf-8') got an unexpected keyword argument 'encoding'
        if isinstance(self.build_fs, LazyFS):
            self.build_fs.wrapped_fs.setcontents(path, v, encoding='utf8')
        else:
            self.build_fs.setcontents(path, v, encoding='utf8')
744534
private void recovery(Long pipelineId) {
        List<Long> currentProcessIds = stageMonitor.getCurrentProcessIds(false);
        for (Long processId : currentProcessIds) {
            String path = StagePathUtils.getProcess(pipelineId, processId);
            try {
                byte[] bytes = zookeeper.readData(path);
                ProcessNodeEventData nodeData = JsonUtils.unmarshalFromByte(bytes, ProcessNodeEventData.class);
                if (nodeData.getStatus().isUnUsed()) {// 加入未使用的processId
                    addReply(processId);
                }
            } catch (ZkException e) {
                logger.error("SelectStageListener", e);
            }
        }
    }
808038
function buildIdToNodeMap(profile) {
        const idToNode = new Map();;
        const stack = [profile.resultRoot];
        //循环遍历
        while (stack.length) {
            const node = stack.pop();
            idToNode.set(node.id, node);
            stack.push.apply(stack, node.children);
        }
        //结果存入 _idToNode 属性
        profile._idToNode = idToNode;
    }
1231477
public List<DataMediaPair> listByPipelineId(Long pipelineId) {
        Assert.assertNotNull(pipelineId);
        List<DataMediaPair> dataMediaPairs = new ArrayList<DataMediaPair>();
        try {
            List<DataMediaPairDO> dataMediaPairDos = dataMediaPairDao.listByPipelineId(pipelineId);
            if (dataMediaPairDos.isEmpty()) {
                logger.debug("DEBUG ## couldn't query any dataMediaPair, maybe hasn't create any dataMediaPair.");
                return dataMediaPairs;
            }
            dataMediaPairs = doToModel(dataMediaPairDos);
        } catch (Exception e) {
            logger.error("ERROR ## query dataMediaPairs by pipelineId:" + pipelineId + " has an exception!", e);
            throw new ManagerException(e);
        }

        return dataMediaPairs;
    }
1274019
@Override
    public MigratablePipeline findPipelineToMigrate(LoadImbalance imbalance) {
        Set<? extends MigratablePipeline> candidates = imbalance.getPipelinesOwnedBy(imbalance.srcOwner);
        long migrationThreshold = (long) ((imbalance.maximumLoad - imbalance.minimumLoad)
                * MAXIMUM_NO_OF_EVENTS_AFTER_MIGRATION_COEFFICIENT);
        MigratablePipeline candidate = null;
        long loadInSelectedPipeline = 0;
        for (MigratablePipeline pipeline : candidates) {
            long load = imbalance.getLoad(pipeline);
            if (load > loadInSelectedPipeline) {
                if (load < migrationThreshold) {
                    loadInSelectedPipeline = load;
                    candidate = pipeline;
                }
            }
        }
        return candidate;
    }
242344
synchronized String serializeNativeFlowContext() { //NOPMD
    	StringBuilder result = new StringBuilder(uniqueId);
        if (showTxCounter) {
            if (innerTxCounter.get() != 0) {
                result = result.append(TX_DELIM).append(innerTxCounter);
            }
        }
        return result.toString();
    }