Naive implementation of generating jobSpec, which fetch the first available template,
in an exemplified single-hop FlowCompiler implementation.
@param flowSpec
@return
162092
protected JobSpec jobSpecGenerator(FlowSpec flowSpec) {
    JobSpec jobSpec;
    JobSpec.Builder jobSpecBuilder = JobSpec.builder(jobSpecURIGenerator(flowSpec))
        .withConfig(flowSpec.getConfig())
        .withDescription(flowSpec.getDescription())
        .withVersion(flowSpec.getVersion());

    if (flowSpec.getTemplateURIs().isPresent() && templateCatalog.isPresent()) {
      // Only first template uri will be honored for Identity
      jobSpecBuilder = jobSpecBuilder.withTemplate(flowSpec.getTemplateURIs().get().iterator().next());
      try {
        jobSpec = new ResolvedJobSpec(jobSpecBuilder.build(), templateCatalog.get());
        log.info("Resolved JobSpec properties are: " + jobSpec.getConfigAsProperties());
      } catch (SpecNotFoundException | JobTemplate.TemplateException e) {
        throw new RuntimeException("Could not resolve template in JobSpec from TemplateCatalog", e);
      }
    } else {
      jobSpec = jobSpecBuilder.build();
      log.info("Unresolved JobSpec properties are: " + jobSpec.getConfigAsProperties());
    }

    // Remove schedule
    jobSpec.setConfig(jobSpec.getConfig().withoutPath(ConfigurationKeys.JOB_SCHEDULE_KEY));

    // Add job.name and job.group
    if (flowSpec.getConfig().hasPath(ConfigurationKeys.FLOW_NAME_KEY)) {
      jobSpec.setConfig(jobSpec.getConfig()
          .withValue(ConfigurationKeys.JOB_NAME_KEY, flowSpec.getConfig().getValue(ConfigurationKeys.FLOW_NAME_KEY)));
    }
    if (flowSpec.getConfig().hasPath(ConfigurationKeys.FLOW_GROUP_KEY)) {
      jobSpec.setConfig(jobSpec.getConfig()
          .withValue(ConfigurationKeys.JOB_GROUP_KEY, flowSpec.getConfig().getValue(ConfigurationKeys.FLOW_GROUP_KEY)));
    }

    // Add flow execution id for this compilation
    long flowExecutionId = FlowUtils.getOrCreateFlowExecutionId(flowSpec);
    jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.FLOW_EXECUTION_ID_KEY,
        ConfigValueFactory.fromAnyRef(flowExecutionId)));

    // Reset properties in Spec from Config
    jobSpec.setConfigAsProperties(ConfigUtils.configToProperties(jobSpec.getConfig()));
    return jobSpec;
  }
922831
protected T createAndInitQueue(QueueSpec spec) throws Exception {
        T queue = createQueueInstance(spec);
        queue.setQueueName(spec.name);
        initQueue(queue, spec);
        return queue;
    }
1294858
private synchronized RunningJobImpl get(final String jobIdentifier) {
    final RunningJobImpl result = this.jobs.get(jobIdentifier);
    if (null == result) {
      throw new RuntimeException("Trying to get a RunningJob that is unknown: " + jobIdentifier);
    }
    return result;
  }
797506
public static StepExecutionItem createJobRef(
            final String jobIdentifier,
            final String[] args,
            final boolean nodeStep,
            final StepExecutionItem handler,
            final boolean keepgoingOnSuccess,
            final String nodeFilter,
            final Integer nodeThreadcount,
            final Boolean nodeKeepgoing,
            final String nodeRankAttribute,
            final Boolean nodeRankOrderAscending,
            final String label,
            final Boolean nodeIntersect,
            final String project,
            final Boolean failOnDisable,
            final Boolean importOptions,
            final String uuid,
            final Boolean useName
    )
    {

        return new JobReferenceItem(
                label,
                jobIdentifier,
                args,
                nodeStep,
                handler,
                keepgoingOnSuccess,
                nodeKeepgoing,
                nodeFilter,
                nodeThreadcount,
                nodeRankAttribute,
                nodeRankOrderAscending,
                nodeIntersect,
                project,
                failOnDisable,
                importOptions,
                uuid,
                useName
        );
    }
896830
public Step<JobXMLDescriptor> getOrCreateStep()
   {
      List<Node> nodeList = model.get("step");
      if (nodeList != null &&  nodeList.size() > 0)
      {
         return new StepImpl<JobXMLDescriptor>(this, "step", model, nodeList.get(0));
      }
      return createStep();
   }
1006056
public Object createNewInstance() throws BlueprintTemplateException {
		if ( this.getNewInstance() != null ) {
            // XXX: warn if the return of the callback is not the same class as the target?
			return this.getNewInstance().createInstance();
		} else {
			return this.getTemplate().construct( this.getTarget() );
		}
	}
332869
public JobWithJars getPlanWithoutJars() throws ProgramInvocationException {
		if (isUsingProgramEntryPoint()) {
			return new JobWithJars(getPlan(), Collections.<URL>emptyList(), classpaths, userCodeClassLoader);
		} else {
			throw new ProgramInvocationException("Cannot create a " + JobWithJars.class.getSimpleName() +
				" for a program that is using the interactive mode.", getPlan().getJobId());
		}
	}
371528
public Next<Flow<T>> getOrCreateNext()
   {
      List<Node> nodeList = childNode.get("next");
      if (nodeList != null &&  nodeList.size() > 0)
      {
         return new NextImpl<Flow<T>>(this, "next", childNode, nodeList.get(0));
      }
      return createNext();
   }
1346524
private WorkUnitStream prepareWorkUnits(WorkUnitStream workUnits, JobState jobState) {
    return workUnits.transform(new WorkUnitPreparator(this.jobContext.getJobId()));
  }
1293859
@Bean
  public JobFactory<FileIngestJobExecution> fileIngestJobFactory() {
    return new JobFactory<FileIngestJobExecution>() {
      @Override
      public Job createJob(FileIngestJobExecution fileIngestJobExecution) {
        final String targetEntityId = fileIngestJobExecution.getTargetEntityId();
        final String url = fileIngestJobExecution.getUrl();
        final String loader = fileIngestJobExecution.getLoader();
        String dataExplorerURL = menuReaderService.findMenuItemPath("dataexplorer");
        fileIngestJobExecution.setResultUrl(
            format("{0}?entity={1}", dataExplorerURL, targetEntityId));
        return progress ->
            fileIngester.ingest(
                targetEntityId, url, loader, fileIngestJobExecution.getIdentifier(), progress);
      }
    };
  }
19938
@Override
  public synchronized JobSpec getJobSpec(URI uri)
      throws JobSpecNotFoundException {
    try {
      Path targetJobSpecFullPath = getPathForURI(this.jobConfDirPath, uri);
      return this.converter.apply(loader.loadPullFile(targetJobSpecFullPath, this.sysConfig, shouldLoadGlobalConf()));
    } catch (FileNotFoundException e) {
      throw new JobSpecNotFoundException(uri);
    } catch (IOException e) {
      throw new RuntimeException("IO exception thrown on loading single job configuration file:" + e.getMessage());
    }
  }
1318384
public FacesConfigFlowDefinitionFlowCallType<FacesConfigFlowDefinitionType<T>> getOrCreateFlowCall()
   {
      List<Node> nodeList = childNode.get("flow-call");
      if (nodeList != null &&  nodeList.size() > 0)
      {
         return new FacesConfigFlowDefinitionFlowCallTypeImpl<FacesConfigFlowDefinitionType<T>>(this, "flow-call", childNode, nodeList.get(0));
      }
      return createFlowCall();
   }
258684
@Override
  public GeneratedCode generate(ClassTemplateSpec templateSpec) {

    String code;
    SwiftProperties swiftProperties = globalConfig.lookupSwiftProperties(templateSpec);
    if (swiftProperties.omit) return null;

    SwiftSyntax syntax = new SwiftSyntax(templateSpec, swiftProperties, globalConfig);
    SwiftyJSON swifty = new SwiftyJSON(syntax);

    try {
      if (templateSpec instanceof RecordTemplateSpec) {
        code = engine.render("rythm-swift/record.txt", templateSpec, syntax, swifty);
      } else if (templateSpec instanceof EnumTemplateSpec) {
        code = engine.render("rythm-swift/enum.txt", templateSpec);
      } else if (templateSpec instanceof UnionTemplateSpec) {
        code = engine.render("rythm-swift/union.txt", templateSpec, syntax, swifty);
      } else {
        return null; // Indicates that we are declining to generate code for the type (e.g. map or array)
      }
    } catch (RythmException e) {
      throw new RuntimeException(
        "Internal error in generator while processing " + templateSpec.getFullName(), e);
    }
    SwiftCompilationUnit compilationUnit =
        new SwiftCompilationUnit(
            templateSpec.getClassName(), "");
    code = formatter.format(code);
    return new GeneratedCode(compilationUnit, code);
  }
175617
public static FlowSpec.Builder builder(URI catalogURI, Properties flowProps) {
    String name = flowProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY);
    String group = flowProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY, "default");

    try {
      URI flowURI = new URI(catalogURI.getScheme(), catalogURI.getAuthority(),
          "/" + group + "/" + name, null);
      FlowSpec.Builder builder = new FlowSpec.Builder(flowURI).withConfigAsProperties(flowProps);
      String descr = flowProps.getProperty(ConfigurationKeys.FLOW_DESCRIPTION_KEY, null);
      if (null != descr) {
        builder = builder.withDescription(descr);
      }

      return builder;
    } catch (URISyntaxException e) {
      throw new RuntimeException("Unable to create a FlowSpec URI: " + e, e);
    }
  }
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();
        }
    }
1278973
static String getFullyQualifiedDagName(Dag<JobExecutionPlan> dag) {
    Config jobConfig = dag.getStartNodes().get(0).getValue().getJobSpec().getConfig();
    String flowGroup = ConfigUtils.getString(jobConfig, ConfigurationKeys.FLOW_GROUP_KEY, "");
    String flowName = ConfigUtils.getString(jobConfig, ConfigurationKeys.FLOW_NAME_KEY, "");
    Long flowExecutionId = ConfigUtils.getLong(jobConfig, ConfigurationKeys.FLOW_EXECUTION_ID_KEY, 0L);

    return "(flowGroup: " + flowGroup + ", flowName: " + flowName + ", flowExecutionId: " + flowExecutionId + ")";
  }
426671
private JobTaskVertex createSingleInputVertex(SingleInputPlanNode node) throws CompilerException {
		final String taskName = node.getNodeName();
		final DriverStrategy ds = node.getDriverStrategy();
		
		// check, whether chaining is possible
		boolean chaining = false;
		{
			Channel inConn = node.getInput();
			PlanNode pred = inConn.getSource();
			chaining = ds.getPushChainDriverClass() != null &&
					!(pred instanceof NAryUnionPlanNode) &&	// first op after union is stand-alone, because union is merged
					!(pred instanceof BulkPartialSolutionPlanNode) &&	// partial solution merges anyways
					!(pred instanceof WorksetPlanNode) &&	// workset merges anyways
					!(pred instanceof IterationPlanNode) && // cannot chain with iteration heads currently
					inConn.getShipStrategy() == ShipStrategyType.FORWARD &&
					inConn.getLocalStrategy() == LocalStrategy.NONE &&
					pred.getOutgoingChannels().size() == 1 &&
					node.getDegreeOfParallelism() == pred.getDegreeOfParallelism() && 
					node.getSubtasksPerInstance() == pred.getSubtasksPerInstance() &&
					node.getBroadcastInputs().isEmpty();
			
			// cannot chain the nodes that produce the next workset or the next solution set, if they are not the
			// in a tail 
			if (this.currentIteration != null && this.currentIteration instanceof WorksetIterationPlanNode &&
					node.getOutgoingChannels().size() > 0)
			{
				WorksetIterationPlanNode wspn = (WorksetIterationPlanNode) this.currentIteration;
				if (wspn.getSolutionSetDeltaPlanNode() == pred || wspn.getNextWorkSetPlanNode() == pred) {
					chaining = false;
				}
			}
			// cannot chain the nodes that produce the next workset in a bulk iteration if a termination criterion follows
			if (this.currentIteration != null && this.currentIteration instanceof BulkIterationPlanNode)
			{
				BulkIterationPlanNode wspn = (BulkIterationPlanNode) this.currentIteration;
				if (node == wspn.getRootOfTerminationCriterion() && wspn.getRootOfStepFunction() == pred){
					chaining = false;
				}else if(node.getOutgoingChannels().size() > 0 &&(wspn.getRootOfStepFunction() == pred ||
						wspn.getRootOfTerminationCriterion() == pred)) {
					chaining = false;
				}
			}
		}
		
		final JobTaskVertex vertex;
		final TaskConfig config;
		
		if (chaining) {
			vertex = null;
			config = new TaskConfig(new Configuration());
			this.chainedTasks.put(node, new TaskInChain(ds.getPushChainDriverClass(), config, taskName));
		} else {
			// create task vertex
			vertex = new JobTaskVertex(taskName, this.jobGraph);
			vertex.setTaskClass( (this.currentIteration != null && node.isOnDynamicPath()) ? IterationIntermediatePactTask.class : RegularPactTask.class);
			
			config = new TaskConfig(vertex.getConfiguration());
			config.setDriver(ds.getDriverClass());
		}
		
		// set user code
		config.setStubWrapper(node.getPactContract().getUserCodeWrapper());
		config.setStubParameters(node.getPactContract().getParameters());
		
		// set the driver strategy
		config.setDriverStrategy(ds);
		if (node.getComparator() != null) {
			config.setDriverComparator(node.getComparator(), 0);
		}
		
		// assign memory, file-handles, etc.
		assignDriverResources(node, config);
		return vertex;
	}