IOUtils.copyCharStream(reader, writer);
final int maxRetries = 2;
final int retryInterval = 10;
final DagEngine engine = new DagEngine("u", "a");
Configuration conf = new XConfiguration();
conf.set(OozieClient.APP_PATH, "file://" + getTestCaseDir() + File.separator + "workflow.xml");
conf.set(OozieClient.USER_NAME, getTestUser());
conf.set(OozieClient.LOG_TOKEN, "t");
conf.set("signal-value", "OK");
conf.set("external-status", "ok");
conf.set("error", errorType);
conf.setInt(OozieClient.ACTION_MAX_RETRIES, maxRetries);
conf.setInt(OozieClient.ACTION_RETRY_INTERVAL, retryInterval);
final String jobId = engine.submitJob(conf, true);
int retryCount = 1;
WorkflowActionBean.Status expectedStatus = expStatus1;
int expectedRetryCount = 2;
Thread.sleep(20000);
String aId = null;
final WorkflowStore store = Services.get().get(WorkflowStoreService.class).create();
store.beginTrx();
while (retryCount <= maxRetries) {
List<WorkflowActionBean> actions = store.getActionsForWorkflow(jobId, false);
WorkflowActionBean action = actions.get(0);
aId = action.getId();
assertEquals(expectedStatus, action.getStatus());
assertEquals(expectedRetryCount, action.getRetries());
assertEquals("TEST_ERROR", action.getErrorCode());
assertEquals(expErrorMsg, action.getErrorMessage());
if (action.getRetries() == maxRetries) {
expectedRetryCount = 0;
expectedStatus = expStatus2;
break;
}
else {
expectedRetryCount++;
}
Thread.sleep(retryInterval * 1000);
retryCount++;
}
store.commitTrx();
store.closeTrx();
Thread.sleep(5000);
final String actionId = aId;
waitFor(5000, new Predicate() {
public boolean evaluate() throws Exception {
return (engine.getWorkflowAction(actionId).getStatus() == expStatus2);
}
});
final WorkflowStore store2 = Services.get().get(WorkflowStoreService.class).create();
store2.beginTrx();
WorkflowActionBean action = engine.getWorkflowAction(actionId);
assertEquals("TEST_ERROR", action.getErrorCode());
assertEquals(expErrorMsg, action.getErrorMessage());
assertEquals(expStatus2, action.getStatus());
assertTrue(action.getPending() == false);
assertEquals(WorkflowJob.Status.SUSPENDED, engine.getJob(jobId).getStatus());
store2.commitTrx();
store2.closeTrx();
}