Create a ExecutionStepContextFetcher to execute fetch.

@param pathFlowSid Flow Sid.
@param pathExecutionSid Execution Sid.
@param pathStepSid Step Sid.
@return ExecutionStepContextFetcher capable of executing the fetch
480416
public static ExecutionStepContextFetcher fetcher(final String pathFlowSid, 
                                                      final String pathExecutionSid, 
                                                      final String pathStepSid) {
        return new ExecutionStepContextFetcher(pathFlowSid, pathExecutionSid, pathStepSid);
    }
410291
public static SyncMapItemUpdater updater(final String pathServiceSid, 
                                             final String pathMapSid, 
                                             final String pathKey) {
        return new SyncMapItemUpdater(pathServiceSid, pathMapSid, pathKey);
    }
1234287
public static FunctionVersionFetcher fetcher(final String pathServiceSid, 
                                                 final String pathFunctionSid, 
                                                 final String pathSid) {
        return new FunctionVersionFetcher(pathServiceSid, pathFunctionSid, pathSid);
    }
197928
private void ajaxFetchExecFlowLogs(final HttpServletRequest req,
      final HttpServletResponse resp, final HashMap<String, Object> ret, final User user,
      final ExecutableFlow exFlow) throws ServletException {
    final long startMs = System.currentTimeMillis();
    final Project project = getProjectAjaxByPermission(ret, exFlow.getProjectId(), user, Type.READ);
    if (project == null) {
      return;
    }

    final int offset = this.getIntParam(req, "offset");
    final int length = this.getIntParam(req, "length");

    resp.setCharacterEncoding("utf-8");

    try {
      final LogData data = this.executorManagerAdapter.getExecutableFlowLog(exFlow, offset, length);
      ret.putAll(appendLogData(data, offset));

    } catch (final ExecutorManagerException e) {
      throw new ServletException(e);
    }

    /*
     * We originally consider leverage Drop Wizard's Timer API {@link com.codahale.metrics.Timer}
     * to measure the duration time.
     * However, Timer will result in too many accompanying metrics (e.g., min, max, 99th quantile)
     * regarding one metrics. We decided to use gauge to do that and monitor how it behaves.
     */
    this.webMetrics.setFetchLogLatency(System.currentTimeMillis() - startMs);
  }
23289
public List<Flow> fetchFlowsWithConfig(String cluster, String username,
      String batchDesc, String signature, int limit, String... configProps)
      throws IOException {
    LOG.info(String.format(
        "Fetching last %d matching jobs for cluster=%s, user.name=%s, "
            + "batch.desc=%s, pig.logical.plan.signature=%s",
        limit, cluster, username, batchDesc, signature));

    String configParam = "";
    if (configProps != null && configProps.length > 0) {
      configParam = StringUtil.buildParam("includeConf", configProps);
    }
    String urlString = signature == null
        ? String.format("http://%s/api/v1/flow/%s/%s/%s?limit=%d&%s",
            apiHostname, cluster, username, StringUtil.cleanseToken(batchDesc),
            limit, configParam)
        : String.format("http://%s/api/v1/flow/%s/%s/%s/%s?limit=%d&%s",
            apiHostname, cluster, username, StringUtil.cleanseToken(batchDesc),
            signature, limit, configParam);

    return retrieveFlowsFromURL(urlString);
  }
985725
@Override
    public List<StepExecution> getStepExecutionsTopLevelFromJobExecutionId(final long jobExecutionId) throws NoSuchJobExecutionException {
        final EntityManager em = getPsu().createEntityManager();
        try {
            List<StepExecution> exec = new TranRequest<List<StepExecution>>(em) {
                @Override
                public List<StepExecution> call() throws Exception {
                    TypedQuery<StepExecution> query = em.createNamedQuery(TopLevelStepExecutionEntity.GET_TOP_LEVEL_STEP_EXECUTIONS_BY_JOB_EXEC_SORT_BY_START_TIME_ASC,
                                                                          StepExecution.class);
                    query.setParameter("jobExecId", jobExecutionId);
                    List<StepExecution> result = query.getResultList();
                    if (result == null) {
                        result = new ArrayList<StepExecution>();
                    }
                    // If empty, try to get job execution to generate NoSuchJobExecutionException if unknown id
                    if (result.isEmpty()) {
                        getJobExecution(jobExecutionId);
                    }
                    return result;
                }
            }.runInNewOrExistingGlobalTran();
            return exec;
        } finally {
            em.close();
        }
    }
964289
public static DeploymentFetcher fetcher(final String pathServiceSid, 
                                            final String pathEnvironmentSid, 
                                            final String pathSid) {
        return new DeploymentFetcher(pathServiceSid, pathEnvironmentSid, pathSid);
    }
3928
public ExecutionReference getReference(final int executionId) {
    if (hasExecution(executionId)) {
      return this.queuedFlowMap.get(executionId).getFirst();
    }
    return null;
  }
103876
public static StepExecutionResult wrapDispatcherResult(final DispatcherResult dispatcherResult) {
        final StepExecutionResultImpl result;
        if(dispatcherResult.isSuccess()) {
            result = NodeDispatchStepExecutorResult.success(dispatcherResult);
        }else{
            result = NodeDispatchStepExecutorResult.failure(dispatcherResult,
                                                            null,
                                                            Reason.NodeDispatchFailure,
                                                            "Node dispatch failed");
            //extract failed nodes
            ArrayList<String> nodeNames = new ArrayList<>();
            for (String nodeName : dispatcherResult.getResults().keySet()) {
                NodeStepResult nodeStepResult = dispatcherResult.getResults().get(nodeName);
                if(!nodeStepResult.isSuccess()) {
                    nodeNames.add(nodeName);
                }
            }
            if(!nodeNames.isEmpty()){
                result.getFailureData().put(FAILURE_DATA_FAILED_NODES, StringUtils.join(nodeNames, ","));
            }
        }
        return result;
    }
1350563
public static long getOrCreateFlowExecutionId(FlowSpec spec) {
    long flowExecutionId;
    if (spec.getConfig().hasPath(ConfigurationKeys.FLOW_EXECUTION_ID_KEY)) {
      flowExecutionId = spec.getConfig().getLong(ConfigurationKeys.FLOW_EXECUTION_ID_KEY);
    } else {
      flowExecutionId = System.currentTimeMillis();
    }
    return flowExecutionId;
  }
1222829
@Override
	public CompletableFuture<StackTraceSampleResponse> requestStackTraceSample(
			final ExecutionAttemptID executionAttemptId,
			final int sampleId,
			final int numSamples,
			final Time delayBetweenSamples,
			final int maxStackTraceDepth,
			final Time timeout) {

		final Task task = taskSlotTable.getTask(executionAttemptId);
		if (task == null) {
			return FutureUtils.completedExceptionally(
				new IllegalStateException(String.format("Cannot sample task %s. " +
					"Task is not known to the task manager.", executionAttemptId)));
		}

		final CompletableFuture<List<StackTraceElement[]>> stackTracesFuture = stackTraceSampleService.requestStackTraceSample(
			TaskStackTraceSampleableTaskAdapter.fromTask(task),
			numSamples,
			delayBetweenSamples,
			maxStackTraceDepth);

		return stackTracesFuture.thenApply(stackTraces ->
			new StackTraceSampleResponse(sampleId, executionAttemptId, stackTraces));
	}
411381
public Map<Integer, Pair<ExecutionReference, ExecutableFlow>> fetchUnfinishedFlowsMetadata()
      throws ExecutorManagerException {
    try {
      return this.dbOperator.query(FetchUnfinishedFlowsMetadata.FETCH_UNFINISHED_FLOWS_METADATA,
          new FetchUnfinishedFlowsMetadata());
    } catch (final SQLException e) {
      throw new ExecutorManagerException("Error fetching unfinished flows metadata", e);
    }
  }
1073887
public static ReservationFetcher fetcher(final String pathWorkspaceSid, 
                                             final String pathTaskSid, 
                                             final String pathSid) {
        return new ReservationFetcher(pathWorkspaceSid, pathTaskSid, pathSid);
    }
728974
public ExecutionImpl createExecution(boolean initializeExecutionStartContext) {
    // create the new child execution
    ExecutionImpl createdExecution = newExecution();

    // initialize sequence counter
    createdExecution.setSequenceCounter(getSequenceCounter());

    // manage the bidirectional parent-child relation
    createdExecution.setParent(this);

    // initialize the new execution
    createdExecution.setProcessDefinition(getProcessDefinition());
    createdExecution.setProcessInstance(getProcessInstance());
    createdExecution.setActivity(getActivity());

    // make created execution start in same activity instance
    createdExecution.activityInstanceId = activityInstanceId;

    // with the fix of CAM-9249 we presume that the parent and the child have the same startContext
    if (initializeExecutionStartContext) {
      createdExecution.setStartContext(new ExecutionStartContext());
    } else if (startContext != null) {
      createdExecution.setStartContext(startContext);
    }

    createdExecution.skipCustomListeners = this.skipCustomListeners;
    createdExecution.skipIoMapping = this.skipIoMapping;

    return createdExecution;
  }
1159301
@Override
    public final void pagedExecute(String queryId, LogicalWorkflow workflow, IResultHandler resultHandler,
                                   int pageSize) throws ConnectorException {
        Long time = null;
        try {
            for (LogicalStep project : workflow.getInitialSteps()) {
                ClusterName clusterName = ((Project) project).getClusterName();
                connectionHandler.startJob(clusterName.getName());
            }
            if (logger.isDebugEnabled()) {
                logger.debug("Async paged Executing [" + workflow.toString() + "] : queryId [" + queryId + "]");
            }
            time = System.currentTimeMillis();
            pagedExecuteWorkFlow(queryId, workflow, resultHandler, pageSize);

            if (logger.isDebugEnabled()) {
                logger.debug(
                        "The async query [" + queryId + "] has ended");
            }
        } finally {
            for (LogicalStep project : workflow.getInitialSteps()) {
                connectionHandler.endJob(((Project) project).getClusterName().getName());
            }
            if (time != null) {
                logger.info("TIME - The execute time to paged executed with queryId [" + queryId + "] has been [" + (System
                        .currentTimeMillis() - time)
                        + "]");
            }
        }

    }
1001891
@Trivial
    public String getStepExecutionCreatedMessage(RuntimeStepExecution runtimeStepExecution) {

        StringBuilder buf = new StringBuilder("\n");
        buf.append(LogHelper.dashes);

        buf.append("For step name = ");
        buf.append(runtimeStepExecution.getStepName());
        buf.append("\n New top-level step execution id = ");
        buf.append(runtimeStepExecution.getInternalStepThreadExecutionId());
        buf.append("\n" + LogHelper.dashes);

        return buf.toString();
    }
554409
public static SyncMapItemFetcher fetcher(final String pathServiceSid, 
                                             final String pathMapSid, 
                                             final String pathKey) {
        return new SyncMapItemFetcher(pathServiceSid, pathMapSid, pathKey);
    }