/* (non-Javadoc)
@see com.ibm.jbatch.container.services.IPersistenceManagerService#createJobExecution(com.ibm.jbatch.container.jsl.JobNavigator, javax.batch.runtime.JobInstance, java.util.Properties, com.ibm.jbatch.container.context.impl.JobContextImpl)
979494
@Override
	public RuntimeJobExecution createJobExecution(JobInstance jobInstance, Properties jobParameters, BatchStatus batchStatus) {
		Timestamp now = new Timestamp(System.currentTimeMillis());
		long newExecutionId = createRuntimeJobExecutionEntry(jobInstance, jobParameters, batchStatus, now);
		RuntimeJobExecution jobExecution = new RuntimeJobExecution(jobInstance, newExecutionId);
		jobExecution.setBatchStatus(batchStatus.name());
		jobExecution.setCreateTime(now);
		jobExecution.setLastUpdateTime(now);
		return jobExecution;
	}
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;
  }
94905
public static TimerJobEntity createTimerEntityForTimerEventDefinition(TimerEventDefinition timerEventDefinition, boolean isInterruptingTimer, 
      ExecutionEntity executionEntity, String jobHandlerType, String jobHandlerConfig) {

    ProcessEngineConfigurationImpl processEngineConfiguration = Context.getProcessEngineConfiguration();
    
    String businessCalendarRef = null;
    Expression expression = null;
    ExpressionManager expressionManager = processEngineConfiguration.getExpressionManager();
    
    // ACT-1415: timer-declaration on start-event may contain expressions NOT
    // evaluating variables but other context, evaluating should happen nevertheless
    VariableScope scopeForExpression = executionEntity;
    if (scopeForExpression == null) {
      scopeForExpression = NoExecutionVariableScope.getSharedInstance();
    }
    
    if (StringUtils.isNotEmpty(timerEventDefinition.getTimeDate())) {

      businessCalendarRef = DueDateBusinessCalendar.NAME;
      expression = expressionManager.createExpression(timerEventDefinition.getTimeDate());

    } else if (StringUtils.isNotEmpty(timerEventDefinition.getTimeCycle())) {

      businessCalendarRef = CycleBusinessCalendar.NAME;
      expression = expressionManager.createExpression(timerEventDefinition.getTimeCycle());

    } else if (StringUtils.isNotEmpty(timerEventDefinition.getTimeDuration())) {

      businessCalendarRef = DurationBusinessCalendar.NAME;
      expression = expressionManager.createExpression(timerEventDefinition.getTimeDuration());
    }
    
    if (StringUtils.isNotEmpty(timerEventDefinition.getCalendarName())) {
      businessCalendarRef = timerEventDefinition.getCalendarName();
      Expression businessCalendarExpression = expressionManager.createExpression(businessCalendarRef);
      businessCalendarRef = businessCalendarExpression.getValue(scopeForExpression).toString();
    }

    if (expression == null) {
      throw new ActivitiException("Timer needs configuration (either timeDate, timeCycle or timeDuration is needed) (" + timerEventDefinition.getId() + ")");
    }

    BusinessCalendar businessCalendar = processEngineConfiguration.getBusinessCalendarManager().getBusinessCalendar(businessCalendarRef);

    String dueDateString = null;
    Date duedate = null;
    
    Object dueDateValue = expression.getValue(scopeForExpression);
    if (dueDateValue instanceof String) {
      dueDateString = (String) dueDateValue;
      
    } else if (dueDateValue instanceof Date) {
      duedate = (Date) dueDateValue;
      
    } else if (dueDateValue instanceof DateTime) {
      //JodaTime support
      duedate = ((DateTime) dueDateValue).toDate();
      
    } else if (dueDateValue != null) {
      throw new ActivitiException("Timer '" + executionEntity.getActivityId()
          + "' was not configured with a valid duration/time, either hand in a java.util.Date or a String in format 'yyyy-MM-dd'T'hh:mm:ss'");
    }
    
    if (duedate == null && dueDateString != null) {
      duedate = businessCalendar.resolveDuedate(dueDateString);
    }

    TimerJobEntity timer = null;
    if (duedate != null) {
      timer = Context.getCommandContext().getTimerJobEntityManager().create();
      timer.setJobType(JobEntity.JOB_TYPE_TIMER);
      timer.setRevision(1);
      timer.setJobHandlerType(jobHandlerType);
      timer.setJobHandlerConfiguration(jobHandlerConfig);
      timer.setExclusive(true);
      timer.setRetries(processEngineConfiguration.getAsyncExecutorNumberOfRetries());
      timer.setDuedate(duedate);
      if (executionEntity != null) {
        timer.setExecution(executionEntity);
        timer.setProcessDefinitionId(executionEntity.getProcessDefinitionId());
        timer.setProcessInstanceId(executionEntity.getProcessInstanceId());

        // Inherit tenant identifier (if applicable)
        if (executionEntity.getTenantId() != null) {
          timer.setTenantId(executionEntity.getTenantId());
        }
      }
    }

    if (StringUtils.isNotEmpty(timerEventDefinition.getTimeCycle())) {
      // See ACT-1427: A boundary timer with a cancelActivity='true', doesn't need to repeat itself
      boolean repeat = !isInterruptingTimer;

      // ACT-1951: intermediate catching timer events shouldn't repeat according to spec
      if (executionEntity != null) {
        FlowElement currentElement = executionEntity.getCurrentFlowElement();
        if (currentElement instanceof IntermediateCatchEvent) {
          repeat = false;
        }
      }

      if (repeat) {
        String prepared = prepareRepeat(dueDateString);
        timer.setRepeat(prepared);
      }
    }

    if (timer != null && executionEntity != null) {
      timer.setExecution(executionEntity);
      timer.setProcessDefinitionId(executionEntity.getProcessDefinitionId());

      // Inherit tenant identifier (if applicable)
      if (executionEntity.getTenantId() != null) {
        timer.setTenantId(executionEntity.getTenantId());
      }
    }

    return timer;
  }
554784
@Override
    public WSJobExecution updateJobExecutionAndInstanceNotSetToServerYet(
                                                                         long jobExecutionId, Date date) throws ExecutionAssignedToServerException {
        return updateJobExecutionAndInstanceOnStopBeforeServerAssigned(jobExecutionId, date);
    }
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
        );
    }
808031
private void runJobExecutionLauncher() throws JobException {
    long startTime = 0;
    String newPlanningId;
    String jobName = jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY);
    Closer closer = Closer.create();
    try {
      HelixManager planningJobManager = this.taskDriverHelixManager.isPresent()?
          this.taskDriverHelixManager.get() : this.jobHelixManager;

      String builderStr = jobProps.getProperty(GobblinClusterConfigurationKeys.DISTRIBUTED_JOB_LAUNCHER_BUILDER,
          GobblinHelixDistributeJobExecutionLauncher.Builder.class.getName());

      // Check if any existing planning job is running
      Optional<String> planningJobIdFromStore = jobsMapping.getPlanningJobId(jobName);
      boolean nonblocking = false;
      // start of critical section to check if a job with same job name is running
      Lock jobLock = locks.get(jobName);
      jobLock.lock();

      try {
        if (planningJobIdFromStore.isPresent()) {
          String previousPlanningJobId = planningJobIdFromStore.get();

          if (HelixUtils.isJobFinished(previousPlanningJobId, previousPlanningJobId, planningJobManager)) {
            log.info("Previous planning job {} has reached to the final state. Start a new one.", previousPlanningJobId);
          } else {
            boolean killDuplicateJob = PropertiesUtils
                .getPropAsBoolean(this.jobProps, GobblinClusterConfigurationKeys.KILL_DUPLICATE_PLANNING_JOB, String.valueOf(GobblinClusterConfigurationKeys.DEFAULT_KILL_DUPLICATE_PLANNING_JOB));

            if (!killDuplicateJob) {
              log.info("Previous planning job {} has not finished yet. Skip this job.", previousPlanningJobId);
              return;
            } else {
              log.info("Previous planning job {} has not finished yet. Kill it.", previousPlanningJobId);
              long timeOut = PropertiesUtils.getPropAsLong(sysProps, GobblinClusterConfigurationKeys.HELIX_WORKFLOW_DELETE_TIMEOUT_SECONDS,
                  GobblinClusterConfigurationKeys.DEFAULT_HELIX_WORKFLOW_DELETE_TIMEOUT_SECONDS) * 1000;
              try {
                HelixUtils.deleteWorkflow(previousPlanningJobId, planningJobManager, timeOut);
              } catch (HelixException e) {
                log.info("Helix cannot delete previous planning job id {} within {} seconds.", previousPlanningJobId,
                    timeOut / 1000);
                throw new JobException("Helix cannot delete previous planning job id " + previousPlanningJobId, e);
              }
            }
          }
        } else {
          log.info("Planning job for {} does not exist. First time run.", jobName);
        }

        GobblinHelixDistributeJobExecutionLauncher.Builder builder = GobblinConstructorUtils.<GobblinHelixDistributeJobExecutionLauncher.Builder>invokeLongestConstructor(
            new ClassAliasResolver(GobblinHelixDistributeJobExecutionLauncher.Builder.class).resolveClass(builderStr));

        // Make a separate copy because we could update some of attributes in job properties (like adding planning id).
        Properties jobPlanningProps = new Properties();
        jobPlanningProps.putAll(this.jobProps);

        // Inject planning id and start time
        newPlanningId = HelixJobsMapping.createPlanningJobId(jobPlanningProps);
        jobPlanningProps.setProperty(GobblinClusterConfigurationKeys.PLANNING_ID_KEY, newPlanningId);
        jobPlanningProps.setProperty(GobblinClusterConfigurationKeys.PLANNING_JOB_CREATE_TIME, String.valueOf(System.currentTimeMillis()));

        builder.setSysProps(this.sysProps);
        builder.setJobPlanningProps(jobPlanningProps);
        builder.setJobHelixManager(this.jobHelixManager);
        builder.setTaskDriverHelixManager(this.taskDriverHelixManager);
        builder.setAppWorkDir(this.appWorkDir);
        builder.setJobsMapping(this.jobsMapping);
        builder.setPlanningJobLauncherMetrics(this.planningJobLauncherMetrics);
        builder.setHelixMetrics(this.helixMetrics);

        // if the distributed job launcher should wait for planning job completion

        Config combined = ConfigUtils.propertiesToConfig(jobPlanningProps)
            .withFallback(ConfigUtils.propertiesToConfig(sysProps));

        nonblocking = ConfigUtils.getBoolean(combined, GobblinClusterConfigurationKeys.NON_BLOCKING_PLANNING_JOB_ENABLED,
            GobblinClusterConfigurationKeys.DEFAULT_NON_BLOCKING_PLANNING_JOB_ENABLED);

        log.info("Planning job {} started.", newPlanningId);
        GobblinHelixDistributeJobExecutionLauncher launcher = builder.build();
        closer.register(launcher);
        this.jobsMapping.setPlanningJobId(jobName, newPlanningId);
        startTime = System.currentTimeMillis();
        this.currentJobMonitor = launcher.launchJob(null);

        // make sure the planning job is initialized (or visible) to other parallel running threads,
        // so that the same critical section check (querying Helix for job completeness)
        // can be applied.
        HelixUtils.waitJobInitialization(planningJobManager, newPlanningId, newPlanningId, 300_000);

      } finally {
        // end of the critical section to check if a job with same job name is running
        jobLock.unlock();
      }

      // we can remove the job spec from the catalog because Helix will drive this job to the end.
      this.deleteJobSpec();

      // If we are using non-blocking mode, this get() only guarantees the plannning job is submitted.
      // It doesn't guarantee the job will finish because internally we won't wait for Helix completion.
      this.currentJobMonitor.get();
      this.currentJobMonitor = null;
      if (nonblocking) {
        log.info("Planning job {} submitted successfully.", newPlanningId);
      } else {
        log.info("Planning job {} finished.", newPlanningId);
        this.planningJobLauncherMetrics.updateTimeForCompletedPlanningJobs(startTime);
      }
    } catch (Exception e) {
      if (startTime != 0) {
        this.planningJobLauncherMetrics.updateTimeForFailedPlanningJobs(startTime);
      }
      log.error("Failed to run planning job {}", jobName, e);
      throw new JobException("Failed to run planning job " + jobName, e);
    } finally {
      try {
        closer.close();
      } catch (IOException e) {
        throw new JobException("Cannot properly close planning job " + jobName, e);
      }
    }
  }
940796
@Override
  public JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis,
      Configuration jobConf) {
    String appId = getAppId(jobConf);
    String version = jobConf.get(Constants.PIG_VERSION_CONF_KEY,
        Constants.UNKNOWN);
    long pigSubmitTimeMillis = jobConf.getLong(Constants.PIG_RUN_CONF_KEY, 0);

    // This means that Constants.PIG_RUN_CONF_KEY was not present (for jobs
    // launched with an older pig version).
    if (pigSubmitTimeMillis == 0) {
      String pigLogfile = jobConf.get(Constants.PIG_LOG_FILE_CONF_KEY);
      if (pigLogfile == null) {
        // Should be rare, but we're seeing this happen occasionally
        // Give up on grouping the jobs within the run together, and treat these as individual runs.
        pigSubmitTimeMillis = submitTimeMillis;
      } else {
        pigSubmitTimeMillis = getScriptStartTimeFromLogfileName(pigLogfile);
      }
    }

    return create(qualifiedJobId, jobConf, appId, version, Framework.PIG,
        pigSubmitTimeMillis);
  }
220963
public JobExecutionInfo toJobExecutionInfo() {
    JobExecutionInfo jobExecutionInfo = new JobExecutionInfo();

    jobExecutionInfo.setJobName(this.jobName);
    jobExecutionInfo.setJobId(this.jobId);
    if (this.startTime > 0) {
      jobExecutionInfo.setStartTime(this.startTime);
    }
    if (this.endTime > 0) {
      jobExecutionInfo.setEndTime(this.endTime);
    }
    jobExecutionInfo.setDuration(this.duration);
    jobExecutionInfo.setState(JobStateEnum.valueOf(this.state.name()));
    jobExecutionInfo.setLaunchedTasks(this.taskCount);
    jobExecutionInfo.setCompletedTasks(this.getCompletedTasks());
    jobExecutionInfo.setLauncherType(getLauncherType());
    if (getTrackingURL().isPresent()) {
      jobExecutionInfo.setTrackingUrl(getTrackingURL().get());
    }

    // Add task execution information
    TaskExecutionInfoArray taskExecutionInfos = new TaskExecutionInfoArray();
    for (TaskState taskState : this.getTaskStates()) {
      taskExecutionInfos.add(taskState.toTaskExecutionInfo());
    }
    jobExecutionInfo.setTaskExecutions(taskExecutionInfos);

    // Add job metrics
    JobMetrics jobMetrics = JobMetrics.get(this);
    MetricArray metricArray = new MetricArray();

    for (Map.Entry<String, ? extends com.codahale.metrics.Metric> entry : jobMetrics.getMetricContext().getCounters()
        .entrySet()) {
      Metric counter = new Metric();
      counter.setGroup(MetricGroup.JOB.name());
      counter.setName(entry.getKey());
      counter.setType(MetricTypeEnum.valueOf(GobblinMetrics.MetricType.COUNTER.name()));
      counter.setValue(Long.toString(((Counter) entry.getValue()).getCount()));
      metricArray.add(counter);
    }

    for (Map.Entry<String, ? extends com.codahale.metrics.Metric> entry : jobMetrics.getMetricContext().getMeters()
        .entrySet()) {
      Metric meter = new Metric();
      meter.setGroup(MetricGroup.JOB.name());
      meter.setName(entry.getKey());
      meter.setType(MetricTypeEnum.valueOf(GobblinMetrics.MetricType.METER.name()));
      meter.setValue(Double.toString(((Meter) entry.getValue()).getMeanRate()));
      metricArray.add(meter);
    }

    for (Map.Entry<String, ? extends com.codahale.metrics.Metric> entry : jobMetrics.getMetricContext().getGauges()
        .entrySet()) {
      Metric gauge = new Metric();
      gauge.setGroup(MetricGroup.JOB.name());
      gauge.setName(entry.getKey());
      gauge.setType(MetricTypeEnum.valueOf(GobblinMetrics.MetricType.GAUGE.name()));
      gauge.setValue(((Gauge<?>) entry.getValue()).getValue().toString());
      metricArray.add(gauge);
    }

    jobExecutionInfo.setMetrics(metricArray);

    // Add job properties
    Map<String, String> jobProperties = Maps.newHashMap();
    for (String name : this.getPropertyNames()) {
      String value = this.getProp(name);
      if (!Strings.isNullOrEmpty(value)) {
        jobProperties.put(name, value);
      }
    }
    jobExecutionInfo.setJobProperties(new StringMap(jobProperties));

    return jobExecutionInfo;
  }
845468
public Job createJob(String query) {
    // [START ]
    Job job = null;
    JobConfiguration jobConfiguration = QueryJobConfiguration.of(query);
    JobInfo jobInfo = JobInfo.of(jobConfiguration);
    try {
      job = bigquery.create(jobInfo);
    } catch (BigQueryException e) {
      // the job was not created
    }
    // [END ]
    return job;
  }
790755
@Override
  public final void setupJob(org.apache.hadoop.mapreduce.JobContext jobContext
                             ) throws IOException {
    setupJob((JobContext) jobContext);
  }
659483
public static JobContext create(AlluxioConfiguration alluxioConf) {
    JobContext context = new JobContext();
    context.init(alluxioConf);
    return context;
  }
1221101
protected Job postJobWithDescription(JSONObject jobDescription, long notBeforeRealtimeElapsed) {
        String jobId = UUID.randomUUID().toString();
        InternalJob job = new InternalJob(jobId, jobDescription, notBeforeRealtimeElapsed);
        return post(job);
    }
321380
@Override
    public WSJobInstance createJobInstance(String appName, String jobXMLName, String jsl, String correlationId) {

        if (authService != null) {
            authService.authorizedJobSubmission();
        }

        return batchKernelService.createJobInstance(appName, jobXMLName, batchSecurityHelper.getRunAsUser(), jsl, correlationId);

    }
1280767
public static ArchivedExecutionGraph createFrom(ExecutionGraph executionGraph) {
		final int numberVertices = executionGraph.getTotalNumberOfVertices();

		Map<JobVertexID, ArchivedExecutionJobVertex> archivedTasks = new HashMap<>(numberVertices);
		List<ArchivedExecutionJobVertex> archivedVerticesInCreationOrder = new ArrayList<>(numberVertices);

		for (ExecutionJobVertex task : executionGraph.getVerticesTopologically()) {
			ArchivedExecutionJobVertex archivedTask = task.archive();
			archivedVerticesInCreationOrder.add(archivedTask);
			archivedTasks.put(task.getJobVertexId(), archivedTask);
		}

		final Map<String, SerializedValue<OptionalFailure<Object>>> serializedUserAccumulators =
			executionGraph.getAccumulatorsSerialized();

		final long[] timestamps = new long[JobStatus.values().length];

		for (JobStatus jobStatus : JobStatus.values()) {
			final int ordinal = jobStatus.ordinal();
			timestamps[ordinal] = executionGraph.getStatusTimestamp(jobStatus);
		}

		return new ArchivedExecutionGraph(
			executionGraph.getJobID(),
			executionGraph.getJobName(),
			archivedTasks,
			archivedVerticesInCreationOrder,
			timestamps,
			executionGraph.getState(),
			executionGraph.getFailureInfo(),
			executionGraph.getJsonPlan(),
			executionGraph.getAccumulatorResultsStringified(),
			serializedUserAccumulators,
			executionGraph.getArchivedExecutionConfig(),
			executionGraph.isStoppable(),
			executionGraph.getCheckpointCoordinatorConfiguration(),
			executionGraph.getCheckpointStatsSnapshot());
	}
336195
@Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC)
    protected void setJobExecutionEndCallbackService(IJobExecutionEndCallbackService reference) {
        jobExecutionEndCallbacks.add(reference);
    }
762321
public List<JobExecutionEntity> getJobExecutionsRunningLocalToServer(PersistenceServiceUnit psu) {

        final EntityManager em = psu.createEntityManager();
        try {
            List<JobExecutionEntity> exec = new TranRequest<List<JobExecutionEntity>>(em) {
                @Override
                public List<JobExecutionEntity> call() throws Exception {
                    TypedQuery<JobExecutionEntity> query = em.createNamedQuery(JobExecutionEntity.GET_JOB_EXECUTIONS_BY_SERVERID_AND_STATUSES_QUERY,
                                                                               JobExecutionEntity.class);
                    query.setParameter("serverid", batchLocationService.getServerId());
                    query.setParameter("status", RUNNING_STATUSES);
                    List<JobExecutionEntity> result = query.getResultList();
                    if (result == null) {
                        return new ArrayList<JobExecutionEntity>();
                    }
                    return result;
                }
            }.runInNewOrExistingGlobalTran();

            return exec;
        } finally {
            em.close();
        }
    }
671737
private Job createJob() {
    Job job = new Job();
    JobConfiguration jobConfig = new JobConfiguration();
    JobConfigurationLoad loadConfig = new JobConfigurationLoad();
    jobConfig.setLoad(loadConfig);
    job.setConfiguration(jobConfig);

    loadConfig.setAllowQuotedNewlines(false);
    loadConfig.setSourceFormat("NEWLINE_DELIMITED_JSON");

    List<String> sources = new ArrayList<String>();
    for (GcsFilename file : fileSet) {
      sources.add(getFileUri(file));
    }

    loadConfig.setSourceUris(sources);

    TableReference tableRef = new TableReference();
    tableRef.setDatasetId(dataset);
    tableRef.setTableId(tableName);
    tableRef.setProjectId(projectId);
    loadConfig.setDestinationTable(tableRef);
    loadConfig.setSchema(schema.getValue());
    return job;
  }