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);
      }
    }
  }
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 ( ) ) ; Optional < String > planningJobIdFromStore = jobsMapping . getPlanningJobId ( jobName ) ; boolean nonblocking = false ; 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 ) ) ; Properties jobPlanningProps = new Properties ( ) ; jobPlanningProps . putAll ( this . jobProps ) ; 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 ) ; 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 ) ; HelixUtils . waitJobInitialization ( planningJobManager , newPlanningId , newPlanningId , 300_000 ) ; } finally { jobLock . unlock ( ) ; } this . deleteJobSpec ( ) ; 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 ) ; } } }
<p> Launch a planning job. The actual job will be launched
on task driver instance, which will handle the early-stop case
by a single while-loop.

@see {@link GobblinHelixJobTask#run()} for the task driver logic.