@Test public void testGetPreviousWorkUnitStatesOnPartialRetryPartialCommit() { SourceState sourceState = new SourceState(new State(), this.previousWorkUnitStates); sourceState.setProp(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY, "onpartial"); sourceState.setProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, "partial"); List<WorkUnitState> returnedWorkUnitStates = this.testSource.getPreviousWorkUnitStatesForRetry(sourceState); Assert.assertEquals(returnedWorkUnitStates, this.expectedPreviousWorkUnitStates); }
protected List<WorkUnitState> getPreviousWorkUnitStatesForRetry(SourceState state) { if (Iterables.isEmpty(state.getPreviousWorkUnitStates())) { return ImmutableList.of(); } WorkUnitRetryPolicy workUnitRetryPolicy; if (state.contains(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)) { workUnitRetryPolicy = WorkUnitRetryPolicy.forName(state.getProp(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)); } else { boolean retryFailedWorkUnits = state.getPropAsBoolean(ConfigurationKeys.WORK_UNIT_RETRY_ENABLED_KEY, true); workUnitRetryPolicy = retryFailedWorkUnits ? WorkUnitRetryPolicy.ALWAYS : WorkUnitRetryPolicy.NEVER; } if (workUnitRetryPolicy == WorkUnitRetryPolicy.NEVER) { return ImmutableList.of(); } List<WorkUnitState> previousWorkUnitStates = Lists.newArrayList(); for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) { if (workUnitState.getWorkingState() != WorkUnitState.WorkingState.COMMITTED) { if (state.getPropAsBoolean(ConfigurationKeys.OVERWRITE_CONFIGS_IN_STATESTORE, ConfigurationKeys.DEFAULT_OVERWRITE_CONFIGS_IN_STATESTORE)) { WorkUnitState workUnitStateCopy = new WorkUnitState(workUnitState.getWorkunit(), state); workUnitStateCopy.addAll(workUnitState); workUnitStateCopy.overrideWith(state); previousWorkUnitStates.add(workUnitStateCopy); } else { previousWorkUnitStates.add(workUnitState); } } } if (workUnitRetryPolicy == WorkUnitRetryPolicy.ALWAYS) { return previousWorkUnitStates; } JobCommitPolicy jobCommitPolicy = JobCommitPolicy .forName(state.getProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY)); if ((workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_PARTIAL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_PARTIAL_SUCCESS) || (workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_FULL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_FULL_SUCCESS)) { return previousWorkUnitStates; } return ImmutableList.of(); }
AbstractSource implements Source<S, D> { protected List<WorkUnitState> getPreviousWorkUnitStatesForRetry(SourceState state) { if (Iterables.isEmpty(state.getPreviousWorkUnitStates())) { return ImmutableList.of(); } WorkUnitRetryPolicy workUnitRetryPolicy; if (state.contains(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)) { workUnitRetryPolicy = WorkUnitRetryPolicy.forName(state.getProp(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)); } else { boolean retryFailedWorkUnits = state.getPropAsBoolean(ConfigurationKeys.WORK_UNIT_RETRY_ENABLED_KEY, true); workUnitRetryPolicy = retryFailedWorkUnits ? WorkUnitRetryPolicy.ALWAYS : WorkUnitRetryPolicy.NEVER; } if (workUnitRetryPolicy == WorkUnitRetryPolicy.NEVER) { return ImmutableList.of(); } List<WorkUnitState> previousWorkUnitStates = Lists.newArrayList(); for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) { if (workUnitState.getWorkingState() != WorkUnitState.WorkingState.COMMITTED) { if (state.getPropAsBoolean(ConfigurationKeys.OVERWRITE_CONFIGS_IN_STATESTORE, ConfigurationKeys.DEFAULT_OVERWRITE_CONFIGS_IN_STATESTORE)) { WorkUnitState workUnitStateCopy = new WorkUnitState(workUnitState.getWorkunit(), state); workUnitStateCopy.addAll(workUnitState); workUnitStateCopy.overrideWith(state); previousWorkUnitStates.add(workUnitStateCopy); } else { previousWorkUnitStates.add(workUnitState); } } } if (workUnitRetryPolicy == WorkUnitRetryPolicy.ALWAYS) { return previousWorkUnitStates; } JobCommitPolicy jobCommitPolicy = JobCommitPolicy .forName(state.getProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY)); if ((workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_PARTIAL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_PARTIAL_SUCCESS) || (workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_FULL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_FULL_SUCCESS)) { return previousWorkUnitStates; } return ImmutableList.of(); } }
AbstractSource implements Source<S, D> { protected List<WorkUnitState> getPreviousWorkUnitStatesForRetry(SourceState state) { if (Iterables.isEmpty(state.getPreviousWorkUnitStates())) { return ImmutableList.of(); } WorkUnitRetryPolicy workUnitRetryPolicy; if (state.contains(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)) { workUnitRetryPolicy = WorkUnitRetryPolicy.forName(state.getProp(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)); } else { boolean retryFailedWorkUnits = state.getPropAsBoolean(ConfigurationKeys.WORK_UNIT_RETRY_ENABLED_KEY, true); workUnitRetryPolicy = retryFailedWorkUnits ? WorkUnitRetryPolicy.ALWAYS : WorkUnitRetryPolicy.NEVER; } if (workUnitRetryPolicy == WorkUnitRetryPolicy.NEVER) { return ImmutableList.of(); } List<WorkUnitState> previousWorkUnitStates = Lists.newArrayList(); for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) { if (workUnitState.getWorkingState() != WorkUnitState.WorkingState.COMMITTED) { if (state.getPropAsBoolean(ConfigurationKeys.OVERWRITE_CONFIGS_IN_STATESTORE, ConfigurationKeys.DEFAULT_OVERWRITE_CONFIGS_IN_STATESTORE)) { WorkUnitState workUnitStateCopy = new WorkUnitState(workUnitState.getWorkunit(), state); workUnitStateCopy.addAll(workUnitState); workUnitStateCopy.overrideWith(state); previousWorkUnitStates.add(workUnitStateCopy); } else { previousWorkUnitStates.add(workUnitState); } } } if (workUnitRetryPolicy == WorkUnitRetryPolicy.ALWAYS) { return previousWorkUnitStates; } JobCommitPolicy jobCommitPolicy = JobCommitPolicy .forName(state.getProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY)); if ((workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_PARTIAL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_PARTIAL_SUCCESS) || (workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_FULL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_FULL_SUCCESS)) { return previousWorkUnitStates; } return ImmutableList.of(); }  }
AbstractSource implements Source<S, D> { protected List<WorkUnitState> getPreviousWorkUnitStatesForRetry(SourceState state) { if (Iterables.isEmpty(state.getPreviousWorkUnitStates())) { return ImmutableList.of(); } WorkUnitRetryPolicy workUnitRetryPolicy; if (state.contains(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)) { workUnitRetryPolicy = WorkUnitRetryPolicy.forName(state.getProp(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)); } else { boolean retryFailedWorkUnits = state.getPropAsBoolean(ConfigurationKeys.WORK_UNIT_RETRY_ENABLED_KEY, true); workUnitRetryPolicy = retryFailedWorkUnits ? WorkUnitRetryPolicy.ALWAYS : WorkUnitRetryPolicy.NEVER; } if (workUnitRetryPolicy == WorkUnitRetryPolicy.NEVER) { return ImmutableList.of(); } List<WorkUnitState> previousWorkUnitStates = Lists.newArrayList(); for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) { if (workUnitState.getWorkingState() != WorkUnitState.WorkingState.COMMITTED) { if (state.getPropAsBoolean(ConfigurationKeys.OVERWRITE_CONFIGS_IN_STATESTORE, ConfigurationKeys.DEFAULT_OVERWRITE_CONFIGS_IN_STATESTORE)) { WorkUnitState workUnitStateCopy = new WorkUnitState(workUnitState.getWorkunit(), state); workUnitStateCopy.addAll(workUnitState); workUnitStateCopy.overrideWith(state); previousWorkUnitStates.add(workUnitStateCopy); } else { previousWorkUnitStates.add(workUnitState); } } } if (workUnitRetryPolicy == WorkUnitRetryPolicy.ALWAYS) { return previousWorkUnitStates; } JobCommitPolicy jobCommitPolicy = JobCommitPolicy .forName(state.getProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY)); if ((workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_PARTIAL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_PARTIAL_SUCCESS) || (workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_FULL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_FULL_SUCCESS)) { return previousWorkUnitStates; } return ImmutableList.of(); }  Extract createExtract(TableType type, String namespace, String table); @Override boolean isEarlyStopped(); }
AbstractSource implements Source<S, D> { protected List<WorkUnitState> getPreviousWorkUnitStatesForRetry(SourceState state) { if (Iterables.isEmpty(state.getPreviousWorkUnitStates())) { return ImmutableList.of(); } WorkUnitRetryPolicy workUnitRetryPolicy; if (state.contains(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)) { workUnitRetryPolicy = WorkUnitRetryPolicy.forName(state.getProp(ConfigurationKeys.WORK_UNIT_RETRY_POLICY_KEY)); } else { boolean retryFailedWorkUnits = state.getPropAsBoolean(ConfigurationKeys.WORK_UNIT_RETRY_ENABLED_KEY, true); workUnitRetryPolicy = retryFailedWorkUnits ? WorkUnitRetryPolicy.ALWAYS : WorkUnitRetryPolicy.NEVER; } if (workUnitRetryPolicy == WorkUnitRetryPolicy.NEVER) { return ImmutableList.of(); } List<WorkUnitState> previousWorkUnitStates = Lists.newArrayList(); for (WorkUnitState workUnitState : state.getPreviousWorkUnitStates()) { if (workUnitState.getWorkingState() != WorkUnitState.WorkingState.COMMITTED) { if (state.getPropAsBoolean(ConfigurationKeys.OVERWRITE_CONFIGS_IN_STATESTORE, ConfigurationKeys.DEFAULT_OVERWRITE_CONFIGS_IN_STATESTORE)) { WorkUnitState workUnitStateCopy = new WorkUnitState(workUnitState.getWorkunit(), state); workUnitStateCopy.addAll(workUnitState); workUnitStateCopy.overrideWith(state); previousWorkUnitStates.add(workUnitStateCopy); } else { previousWorkUnitStates.add(workUnitState); } } } if (workUnitRetryPolicy == WorkUnitRetryPolicy.ALWAYS) { return previousWorkUnitStates; } JobCommitPolicy jobCommitPolicy = JobCommitPolicy .forName(state.getProp(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, ConfigurationKeys.DEFAULT_JOB_COMMIT_POLICY)); if ((workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_PARTIAL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_PARTIAL_SUCCESS) || (workUnitRetryPolicy == WorkUnitRetryPolicy.ON_COMMIT_ON_FULL_SUCCESS && jobCommitPolicy == JobCommitPolicy.COMMIT_ON_FULL_SUCCESS)) { return previousWorkUnitStates; } return ImmutableList.of(); }  Extract createExtract(TableType type, String namespace, String table); @Override boolean isEarlyStopped();  }