code: protected void createWorkUnits(SourceState state)
        throws IOException {
      createWorkunitsFromPreviousState(state);
      if (this.datasets.isEmpty()) {
        return;
      }
      for (HivePartitionDataset dataset : this.datasets) {
        Optional<String> owner = dataset.getOwner();
        if (workUnitsExceeded()) {
          log.info("Workunits exceeded");
          setJobWatermark(state, dataset.datasetURN());
          return;
        }
        if (!this.policy.shouldPurge(dataset)) {
          continue;
        }
        WorkUnit workUnit = createNewWorkUnit(dataset);
        log.info("Created new work unit with partition " + workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME));
        this.workUnitMap.put(workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME), workUnit);
        this.workUnitsCreatedCount++;
      }
      if (!state.contains(ComplianceConfigurationKeys.HIVE_PURGER_WATERMARK)) {
        this.setJobWatermark(state, ComplianceConfigurationKeys.NO_PREVIOUS_WATERMARK);
      }
    }
docs: This method creates the list of all work units needed for the current execution.
  Fresh work units are created for each partition starting from watermark and failed work units from the
  previous run will be added to the list.