Sets Job Watermark in the SourceState which will be copied to all WorkUnitStates. Job Watermark is a complete partition name.
During next run of this job, fresh work units will be created starting from this partition.
501883
protected void setJobWatermark(SourceState state, String watermark) {
    state.setProp(ComplianceConfigurationKeys.HIVE_PURGER_WATERMARK, watermark);
    log.info("Setting job watermark for the job: " + watermark);
  }
1017636
function(partition) {
                     var self = this;
                     var results = {
                         'id': [],
                         'partition': [],
                         'nameUUID': [],
                         'status': []
                     };
                     self.data.partition.map(function(val, i) { // for each partition
                         if (val.match(partition)) { // if the job is on @partition
                             for (var key in self.data) { // keep all the {'key':'value'} corresponding
                                 results[key].push(self.data[key][i]);
                             }
                         }
                     });
                     return results;
                 }
1162411
private String persistWorkUnit(final Path workUnitFileDir, final WorkUnit workUnit, ParallelRunner stateSerDeRunner)
      throws IOException {
    final StateStore stateStore;
    String workUnitFileName = workUnit.getId();

    if (workUnit instanceof MultiWorkUnit) {
      workUnitFileName += MULTI_WORK_UNIT_FILE_EXTENSION;
      stateStore = stateStores.getMwuStateStore();
    } else {
      workUnitFileName += WORK_UNIT_FILE_EXTENSION;
      stateStore = stateStores.getWuStateStore();
    }

    Path workUnitFile = new Path(workUnitFileDir, workUnitFileName);
    final String fileName = workUnitFile.getName();
    final String storeName = workUnitFile.getParent().getName();
    stateSerDeRunner.submitCallable(new Callable<Void>() {
      @Override
      public Void call() throws Exception {
        stateStore.put(storeName, fileName, workUnit);
        return null;
      }
    }, "Serialize state to store " + storeName + " file " + fileName);

    return workUnitFile.toString();
  }
150685
public void setLWMScn(String source, long scn) {
        WaterMarkEntry e = sourceWaterMarkMap.get(source);
        if (e == null) {
            e = new WaterMarkEntry(source);
            sourceWaterMarkMap.put(source, e);
        }
        e.setLWMScn(scn);
    }
145106
private void initJobDir (SourceState state) throws IOException {
    String tmpBase = state.getProp(MRCompactor.COMPACTION_TMP_DEST_DIR, MRCompactor.DEFAULT_COMPACTION_TMP_DEST_DIR);
    String jobId;

    if (state instanceof JobState) {
      jobId = ((JobState) state).getJobId();
    } else {
      jobId = UUID.randomUUID().toString();
    }

    this.tmpJobDir = new Path (tmpBase, jobId);
    this.fs.mkdirs(this.tmpJobDir);
    state.setProp (MRCompactor.COMPACTION_JOB_DIR, tmpJobDir.toString());
    log.info ("Job dir is created under {}", this.tmpJobDir);
  }
1036896
protected Job createJob() throws IOException {
    Job job = Job.getInstance(new Configuration());
    for (Map.Entry<Object, Object> entry : this.taskContext.getTaskState().getProperties().entrySet()) {
      if (entry.getKey() instanceof String && ((String) entry.getKey()).startsWith(JOB_CONFIGURATION_PREFIX)) {
        String actualKey = ((String) entry.getKey()).substring(JOB_CONFIGURATION_PREFIX.length());
        job.getConfiguration().set(actualKey, (String) entry.getValue());
      }
    }
    return job;
  }
2594
@Override
  public boolean shouldCreateWorkunit(Partition sourcePartition, LongWatermark lowWatermark) {
    // If a whitelist is provided only create workunits for those partitions
    if (!this.partitionsWhitelist.isEmpty()) {
      return this.partitionsWhitelist.contains(sourcePartition.getCompleteName());
    }
    // If no whitelist is set, all partitions of a dataset are backfilled
    return true;
  }
709542
public static WorkUnit create(Extract extract, WatermarkInterval watermarkInterval) {
    return new WorkUnit(null, extract, watermarkInterval);
  }
754987
public void backoffActualHighWatermark() {
    JsonElement lowWatermark = this.workUnit.getLowWatermark();
    if (lowWatermark == null) {
      return;
    }
    setProp(ConfigurationKeys.WORK_UNIT_STATE_ACTUAL_HIGH_WATER_MARK_KEY, lowWatermark.toString());
  }
152664
void markCompletedJob(JobInProgress job) {
    for (TaskInProgress tip : job.getTasks(TaskType.JOB_SETUP)) {
      for (TaskStatus taskStatus : tip.getTaskStatuses()) {
        if (taskStatus.getRunState() != TaskStatus.State.RUNNING &&
            taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING &&
            taskStatus.getRunState() != TaskStatus.State.UNASSIGNED) {
          markCompletedTaskAttempt(taskStatus.getTaskTracker(),
                                   taskStatus.getTaskID());
        }
      }
    }
    for (TaskInProgress tip : job.getTasks(TaskType.MAP)) {
      for (TaskStatus taskStatus : tip.getTaskStatuses()) {
        if (taskStatus.getRunState() != TaskStatus.State.RUNNING &&
            taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING &&
            taskStatus.getRunState() != TaskStatus.State.FAILED_UNCLEAN &&
            taskStatus.getRunState() != TaskStatus.State.KILLED_UNCLEAN &&
            taskStatus.getRunState() != TaskStatus.State.UNASSIGNED) {
          markCompletedTaskAttempt(taskStatus.getTaskTracker(),
                                   taskStatus.getTaskID());
        }
      }
    }
    for (TaskInProgress tip : job.getTasks(TaskType.REDUCE)) {
      for (TaskStatus taskStatus : tip.getTaskStatuses()) {
        if (taskStatus.getRunState() != TaskStatus.State.RUNNING &&
            taskStatus.getRunState() != TaskStatus.State.COMMIT_PENDING &&
            taskStatus.getRunState() != TaskStatus.State.FAILED_UNCLEAN &&
            taskStatus.getRunState() != TaskStatus.State.KILLED_UNCLEAN &&
            taskStatus.getRunState() != TaskStatus.State.UNASSIGNED) {
          markCompletedTaskAttempt(taskStatus.getTaskTracker(),
                                   taskStatus.getTaskID());
        }
      }
    }
  }
865553
public static Path getJobStateFilePath(boolean usingStateStore, Path appWorkPath, String jobId) {
    final Path jobStateFilePath;

    // the state store uses a path of the form workdir/_jobstate/job_id/job_id.job.state while old method stores the file
    // in the app work dir.
    if (usingStateStore) {
      jobStateFilePath = new Path(appWorkPath, GobblinClusterConfigurationKeys.JOB_STATE_DIR_NAME
          + Path.SEPARATOR + jobId + Path.SEPARATOR + jobId + "."
          + AbstractJobLauncher.JOB_STATE_FILE_NAME);

    } else {
      jobStateFilePath = new Path(appWorkPath, jobId + "." + AbstractJobLauncher.JOB_STATE_FILE_NAME);
    }

    log.info("job state file path: " + jobStateFilePath);

    return jobStateFilePath;
  }
554104
public static Optional<CopyableFileWatermarkGenerator> getCopyableFileWatermarkGenerator(State state)
      throws IOException {
    try {
      if (state.contains(WATERMARK_CREATOR)) {
        Class<?> watermarkCreatorClass = Class.forName(state.getProp(WATERMARK_CREATOR));
        return Optional.of((CopyableFileWatermarkGenerator) watermarkCreatorClass.newInstance());
      } else {
        return Optional.absent();
      }
    } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
      throw new IOException("Failed to instantiate watermarkCreator.");
    }
  }
624645
private static void setJobName(Map<String,String> map, String jobName) {
		map.put(JOB_NAME_KEY, jobName);
	}
567719
private void prepareHadoopJob(List<WorkUnit> workUnits) throws IOException {
    TimingEvent mrJobSetupTimer = this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.MR_JOB_SETUP);

    // Add dependent jars/files
    addDependencies(this.job.getConfiguration());

    this.job.setJarByClass(MRJobLauncher.class);
    this.job.setMapperClass(TaskRunner.class);

    // The job is mapper-only
    this.job.setNumReduceTasks(0);

    this.job.setInputFormatClass(GobblinWorkUnitsInputFormat.class);
    this.job.setOutputFormatClass(GobblinOutputFormat.class);
    this.job.setMapOutputKeyClass(NullWritable.class);
    this.job.setMapOutputValueClass(NullWritable.class);

    // Set speculative execution

    this.job.setSpeculativeExecution(isSpeculativeExecutionEnabled(this.jobProps));

    this.job.getConfiguration().set("mapreduce.job.user.classpath.first", "true");

    // Job input path is where input work unit files are stored

    // Prepare job input
    prepareJobInput(workUnits);
    FileInputFormat.addInputPath(this.job, this.jobInputPath);

    // Job output path is where serialized task states are stored
    FileOutputFormat.setOutputPath(this.job, this.jobOutputPath);

    // Serialize source state to a file which will be picked up by the mappers
    serializeJobState(this.fs, this.mrJobDir, this.conf, this.jobContext.getJobState(), this.job);

    if (this.jobProps.containsKey(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)) {
      GobblinWorkUnitsInputFormat.setMaxMappers(this.job,
          Integer.parseInt(this.jobProps.getProperty(ConfigurationKeys.MR_JOB_MAX_MAPPERS_KEY)));
    }

    this.job.getConfiguration().set(GOBBLIN_JOB_INTERRUPT_PATH_KEY, this.interruptPath.toString());

    mrJobSetupTimer.stop();
  }
338463
@Override
  public List<WorkUnit> getWorkunits(SourceState state) {
    List<WorkUnit> workUnits = Lists.newArrayList();
    Config config = ConfigUtils.propertiesToConfig(state.getProperties());
    Config sourceConfig = ConfigUtils.getConfigOrEmpty(config, DATASET_CLEANER_SOURCE_PREFIX);
    List<String> configurationNames = ConfigUtils.getStringList(config, DATASET_CLEANER_CONFIGURATIONS);

    // use a dummy configuration name if none set
    if (configurationNames.isEmpty()) {
      configurationNames = ImmutableList.of("DummyConfig");
    }

    for (String configurationName: configurationNames) {
      WorkUnit workUnit = WorkUnit.createEmpty();

      // specific configuration prefixed by the configuration name has precedence over the source specific configuration
      // and the source specific configuration has precedence over the general configuration
      Config wuConfig = ConfigUtils.getConfigOrEmpty(sourceConfig, configurationName).withFallback(sourceConfig)
          .withFallback(config);

      workUnit.setProps(ConfigUtils.configToProperties(wuConfig), new Properties());
      TaskUtils.setTaskFactoryClass(workUnit, DatasetCleanerTaskFactory.class);
      workUnits.add(workUnit);
    }

    return workUnits;
  }
205690
private boolean canCleanStagingData(JobState jobState)
      throws IOException {
    return this.jobContext.getSemantics() != DeliverySemantics.EXACTLY_ONCE || !this.jobContext.getCommitSequenceStore()
        .get().exists(jobState.getJobName());
  }
762079
private void updateMinPunctuatedWatermark(Watermark nextWatermark) {
		if (nextWatermark.getTimestamp() > maxWatermarkSoFar) {
			long newMin = Long.MAX_VALUE;

			for (KafkaTopicPartitionState<?> state : subscribedPartitionStates) {
				@SuppressWarnings("unchecked")
				final KafkaTopicPartitionStateWithPunctuatedWatermarks<T, KPH> withWatermarksState =
						(KafkaTopicPartitionStateWithPunctuatedWatermarks<T, KPH>) state;

				newMin = Math.min(newMin, withWatermarksState.getCurrentPartitionWatermark());
			}

			// double-check locking pattern
			if (newMin > maxWatermarkSoFar) {
				synchronized (checkpointLock) {
					if (newMin > maxWatermarkSoFar) {
						maxWatermarkSoFar = newMin;
						sourceContext.emitWatermark(new Watermark(newMin));
					}
				}
			}
		}
	}