code: protected void createWorkunitsFromPreviousState(SourceState state) {
      if (this.lowWatermark.equalsIgnoreCase(ComplianceConfigurationKeys.NO_PREVIOUS_WATERMARK)) {
        return;
      }
      if (Iterables.isEmpty(state.getPreviousWorkUnitStates())) {
        return;
      }
      for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) {
        if (workUnitState.getWorkingState() == WorkUnitState.WorkingState.COMMITTED) {
          continue;
        }
        WorkUnit workUnit = workUnitState.getWorkunit();
        Preconditions.checkArgument(workUnit.contains(ComplianceConfigurationKeys.PARTITION_NAME),
            "Older WorkUnit doesn't contain property partition name.");
        int executionAttempts = workUnit.getPropAsInt(ComplianceConfigurationKeys.EXECUTION_ATTEMPTS,
            ComplianceConfigurationKeys.DEFAULT_EXECUTION_ATTEMPTS);
        if (executionAttempts < this.maxWorkUnitExecutionAttempts) {
          Optional<WorkUnit> workUnitOptional =
              createNewWorkUnit(workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME), ++executionAttempts);
          if (!workUnitOptional.isPresent()) {
            continue;
          }
          workUnit = workUnitOptional.get();
          log.info("Revived old Work Unit for partiton " + workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME)
              + " having execution attempt " + workUnit.getProp(ComplianceConfigurationKeys.EXECUTION_ATTEMPTS));
          workUnitMap.put(workUnit.getProp(ComplianceConfigurationKeys.PARTITION_NAME), workUnit);
        }
      }
    }
docs: Add failed work units in a workUnitMap with partition name as Key.
  New work units are created using required configuration from the old work unit.