销毁和释放对应pipelineId的仲裁资源

@param pipelineId
775310
public static void destory(Long pipelineId) {
        Map<Class, Object> resources = cache.remove(pipelineId);
        if (resources != null) {
            Collection collection = resources.values();
            for (Object obj : collection) {
                if (obj instanceof ArbitrateLifeCycle) {
                    ArbitrateLifeCycle lifeCycle = (ArbitrateLifeCycle) obj;
                    lifeCycle.destory();// 调用销毁方法
                }
            }
        }
    }
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;
    }
919531
public void reload(Long processId) {
        try {
            if (logger.isDebugEnabled()) {
                logger.debug("## reload Stage pipeline[{}] process[{}]", getPipelineId(), processId);
            }

            initProcessStage(processId);
        } catch (Exception cause) {
        }
    }
718759
public boolean ensurePipelineVisible(CaseInsensitiveString pipelineToAdd) {
        boolean modified = false;

        for (DashboardFilter f : viewFilters.filters()) {
            modified = modified || f.allowPipeline(pipelineToAdd);
        }

        return modified;
    }
1345916
public void deletePipeline(String pipelineName) {
        DeletePipelineRequest request = new DeletePipelineRequest();
        request.setPipelineName(pipelineName);
        deletePipeline(request);
    }
145
public static String getTerminRoot(Long channelId, Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_TERMIN_ROOT,
            String.valueOf(channelId),
            String.valueOf(pipelineId));
    }
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);
            }
        }
    }
1073662
public String pipelineWrite(String key, DataGenerator dataGenerator, int max_pipe_keys, int min_pipe_keys)
            throws Exception {
        // Create a random key between [0,MAX_PIPE_KEYS]
        int pipe_keys = randomGenerator.nextInt(max_pipe_keys);

        // Make sure that the number of keys in the pipeline are at least
        // MIN_PIPE_KEYS
        pipe_keys = Math.max(min_pipe_keys, pipe_keys);

        DynoJedisPipeline pipeline = this.jedisClient.get().pipelined();
        Map<String, Response<String>> responses = new HashMap<>();

        /**
         * writeSingle returns a single string, so we want to create a
         * StringBuilder to append all the keys in the form "key_n". This is
         * just used to return a single string
         */
        StringBuilder sb = new StringBuilder();
        // Iterate across the number of keys in the pipeline and set
        for (int n = 0; n < pipe_keys; ++n) {
            String nth_key = key + "_" + n;
            sb.append(nth_key);
            Response<String> resp = pipeline.set(key, key + dataGenerator.getRandomValue() + key);
            responses.put(nth_key, resp);
        }
        pipeline.sync();

        return sb.toString();
    }
1236639
public void deletePipeline(DeletePipelineRequest request) {
        checkNotNull(request, "The parameter request should NOT be null.");
        checkStringNotEmpty(request.getPipelineName(), 
                "The parameter pipelineName should NOT be null or empty string.");
        InternalRequest internalRequest =
                createRequest(HttpMethodName.DELETE, request, PIPELINE, request.getPipelineName());

        invokeHttpClient(internalRequest, CreatePipelineResponse.class);
    }
655015
func (s *PutPipelineDefinitionInput) SetPipelineId(v string) *PutPipelineDefinitionInput {
	s.PipelineId = &v
	return s
}
1044140
public List<Channel> listByNodeId(Long nodeId, ChannelStatus... statuses) {
        List<Channel> channels = new ArrayList<Channel>();
        List<Channel> results = new ArrayList<Channel>();
        try {
            List<Pipeline> pipelines = pipelineService.listByNodeId(nodeId);
            List<Long> pipelineIds = new ArrayList<Long>();
            for (Pipeline pipeline : pipelines) {
                pipelineIds.add(pipeline.getId());
            }

            if (pipelineIds.isEmpty()) { // 没有关联任务直接返回
                return channels;
            }

            // 反查对应的channel
            channels = listByPipelineIds(pipelineIds.toArray(new Long[pipelineIds.size()]));
            if (null == statuses || statuses.length == 0) {
                return channels;
            }

            for (Channel channel : channels) {
                for (ChannelStatus status : statuses) {
                    if (channel.getStatus().equals(status)) {
                        results.add(channel);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("ERROR ## list query channel by nodeId:" + nodeId + " has an exception!");
            throw new ManagerException(e);
        }
        return results;
    }
45421
public static String getExtractStage(Long pipelineId, Long processId) {
        return getProcess(pipelineId, processId) + "/" + ArbitrateConstants.NODE_EXTRACTED;
    }
1175743
public List<Channel> listByPipelineIds(Long... pipelineIds) {
        List<Channel> channels = new ArrayList<Channel>();
        try {
            List<Pipeline> pipelines = pipelineService.listByIds(pipelineIds);

            List<Long> channelIds = new ArrayList<Long>();

            for (Pipeline pipeline : pipelines) {
                if (!channelIds.contains(pipeline.getChannelId())) {
                    channelIds.add(pipeline.getChannelId());
                }
            }
            channels = listByIds(channelIds.toArray(new Long[channelIds.size()]));
        } catch (Exception e) {
            logger.error("ERROR ## list query channel by pipelineIds:" + pipelineIds.toString() + " has an exception!");
            throw new ManagerException(e);
        }
        return channels;
    }
761010
public void setPipelines(java.util.Collection<PipelineSummary> pipelines) {
        if (pipelines == null) {
            this.pipelines = null;
            return;
        }

        this.pipelines = new java.util.ArrayList<PipelineSummary>(pipelines);
    }
781695
public List<RemedyIndexEventData> listRemedyIndexs(Long pipelineId) {
        String path = StagePathUtils.getRemedyRoot(pipelineId);
        List<RemedyIndexEventData> datas = new ArrayList<RemedyIndexEventData>();
        try {
            List<String> nodes = zookeeper.getChildren(path);
            for (String node : nodes) {
                RemedyIndexEventData data = RemedyIndexEventData.parseNodeName(node);
                data.setPipelineId(pipelineId);
                datas.add(data);
            }
        } catch (ZkException e) {
            throw new ArbitrateException("listRemedyIndexs", pipelineId.toString(), e);
        }

        Collections.sort(datas, new RemedyIndexComparator()); // 做一下排序
        return datas;
    }
934336
public static String getRemedyRoot(Long channelId, Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_REMEDY_ROOT,
            String.valueOf(channelId),
            String.valueOf(pipelineId));
    }
120537
public ListPipelinesResponse listPipelines(ListPipelinesRequest request) {
        checkNotNull(request, "The parameter request should NOT be null.");
        InternalRequest internalRequest = createRequest(HttpMethodName.GET, request, PIPELINE);
        return invokeHttpClient(internalRequest, ListPipelinesResponse.class);
    }