You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@falcon.apache.org by sa...@apache.org on 2014/09/26 09:32:28 UTC
git commit: FALCON-645 add test to falcon regression for change made
in default behaviour of instance request contributed by Karishma
Repository: incubator-falcon
Updated Branches:
refs/heads/master 4da62d1c5 -> 94eb546bf
FALCON-645 add test to falcon regression for change made in default behaviour of instance request contributed by Karishma
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/94eb546b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/94eb546b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/94eb546b
Branch: refs/heads/master
Commit: 94eb546bf420ad4809460a7f49ab85b2bed12c9c
Parents: 4da62d1
Author: Samarth Gupta <sa...@inmobi.com>
Authored: Fri Sep 26 13:02:06 2014 +0530
Committer: Samarth Gupta <sa...@inmobi.com>
Committed: Fri Sep 26 13:02:06 2014 +0530
----------------------------------------------------------------------
falcon-regression/CHANGES.txt | 4 +
.../regression/ProcessInstanceKillsTest.java | 81 +++++++------
.../regression/ProcessInstanceRerunTest.java | 53 +++++----
.../regression/ProcessInstanceResumeTest.java | 91 ++++++--------
.../regression/ProcessInstanceRunningTest.java | 35 ++++--
.../regression/ProcessInstanceStatusTest.java | 118 +++++++++----------
.../regression/ProcessInstanceSuspendTest.java | 96 +++++++--------
7 files changed, 241 insertions(+), 237 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/CHANGES.txt
----------------------------------------------------------------------
diff --git a/falcon-regression/CHANGES.txt b/falcon-regression/CHANGES.txt
index 946699f..84580d4 100644
--- a/falcon-regression/CHANGES.txt
+++ b/falcon-regression/CHANGES.txt
@@ -9,6 +9,10 @@ Trunk (Unreleased)
via Samarth Gupta)
IMPROVEMENTS
+
+ FALCON-645 add test to falcon regression for change made in default behaviour
+ of instance request ( Karishma Via Samarth Gupta)
+
FALCON-745 LogUtil requires fine control over what to dump (Raghav Kumar
Gautam)
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceKillsTest.java
----------------------------------------------------------------------
diff --git a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceKillsTest.java b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceKillsTest.java
index 9b8eb30..aa9773c 100644
--- a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceKillsTest.java
+++ b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceKillsTest.java
@@ -24,16 +24,18 @@ import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.InstancesResult;
import org.apache.falcon.regression.core.response.InstancesResult.WorkflowStatus;
-import org.apache.falcon.regression.core.util.BundleUtil;
+import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.core.util.HadoopUtil;
-import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.BundleUtil;
import org.apache.falcon.regression.core.util.OSUtil;
-import org.apache.falcon.regression.core.util.TimeUtil;
+import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.OozieUtil;
import org.apache.falcon.regression.core.util.Util;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.Logger;
import org.apache.oozie.client.CoordinatorAction;
+import org.apache.oozie.client.OozieClient;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -43,7 +45,6 @@ import org.testng.annotations.Test;
import java.io.IOException;
import java.lang.reflect.Method;
-import java.util.List;
/**
* Process instance kill tests.
@@ -52,6 +53,7 @@ import java.util.List;
public class ProcessInstanceKillsTest extends BaseTestClass {
private ColoHelper cluster = servers.get(0);
+ private OozieClient clusterOC = serverOC.get(0);
private FileSystem clusterFS = serverFS.get(0);
private String testDir = "/ProcessInstanceKillsTest";
private String baseTestHDFSDir = baseHDFSDir + testDir;
@@ -59,7 +61,6 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
private String feedInputPath = baseTestHDFSDir + "/input" + MINUTE_DATE_PATTERN;
private String feedOutputPath = baseTestHDFSDir + "/output-data" + MINUTE_DATE_PATTERN;
private static final Logger LOGGER = Logger.getLogger(ProcessInstanceKillsTest.class);
- private static final double TIMEOUT = 15;
private String processName;
@BeforeClass(alwaysRun = true)
@@ -69,12 +70,7 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
Bundle b = BundleUtil.readELBundle();
b.generateUniqueBundle();
b = new Bundle(b, cluster);
- String startDate = "2010-01-01T23:20Z";
- String endDate = "2010-01-02T01:21Z";
b.setInputFeedDataPath(feedInputPath);
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, endDate, 20);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.NORMAL_INPUT,
- b.getFeedDataPathPrefix(), dataDates);
}
@BeforeMethod(alwaysRun = true)
@@ -108,16 +104,17 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:04Z");
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstancesResult r = prism.getProcessHelper().getProcessInstanceKill(processName,
- "?start=2010-01-02T01:00Z");
+ "?start=2010-01-02T01:00Z");
InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.KILLED);
}
/**
* Schedule process. Check that in case when -start and -end parameters are equal -kill
* action results in the same way as in case with only -start parameter is used. Only one
- * instance should be killed.
+ * instance should be killed. This test will fail because of https://issues.apache.org/jira/browse/FALCON-704
*
* @throws Exception
*/
@@ -129,9 +126,12 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
bundles[0].setProcessPeriodicity(1, TimeUnit.minutes);
bundles[0].setProcessConcurrency(10);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 2,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper()
- .getProcessInstanceKill(processName, "?start=2010-01-02T00:03Z&end=2010-01-02T00:03Z");
+ .getProcessInstanceKill(processName, "?start=2010-01-02T00:03Z&end=2010-01-02T00:03Z");
InstanceUtil.validateResponse(r, 1, 0, 0, 0, 1);
}
@@ -143,15 +143,18 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
* @throws Exception
*/
@Test(groups = {"singleCluster"})
- public void testProcessInstanceKillKillNonMatrelized() throws Exception {
- bundles[0].setProcessValidity("2010-01-02T00:00Z", "2010-01-02T04:00Z");
+ public void testProcessInstanceKillKillNonMaterialized() throws Exception {
+ bundles[0].setProcessValidity("2010-01-02T00:00Z", "2010-01-02T01:00Z");
bundles[0].setProcessTimeOut(3, TimeUnit.minutes);
bundles[0].setProcessPeriodicity(1, TimeUnit.minutes);
bundles[0].setProcessConcurrency(6);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper()
- .getProcessInstanceKill(processName, "?start=2010-01-02T00:03Z&end=2010-01-02T00:30Z");
+ .getProcessInstanceKill(processName, "?start=2010-01-02T00:03Z&end=2010-01-02T00:30Z");
InstanceUtil.validateResponse(r, 3, 0, 0, 0, 3);
LOGGER.info(r.toString());
}
@@ -172,17 +175,16 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
String startTimeData = TimeUtil.getTimeWrtSystemTime(-50);
String endTimeData = TimeUtil.getTimeWrtSystemTime(50);
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startTimeData, endTimeData, 1);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.SINGLE_FILE,
- baseTestHDFSDir + "/input01", dataDates);
bundles[0].setInputFeedDataPath(feedInputPath.replace("input/", "input01/"));
bundles[0].setProcessValidity(startTime, endTime);
bundles[0].setProcessConcurrency(6);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
String startTimeRequest = TimeUtil.getTimeWrtSystemTime(-17);
String endTimeRequest = TimeUtil.getTimeWrtSystemTime(23);
InstancesResult r = prism.getProcessHelper().getProcessInstanceKill(processName,
- "?start=" + startTimeRequest + "&end=" + endTimeRequest);
+ "?start=" + startTimeRequest + "&end=" + endTimeRequest);
LOGGER.info(r.toString());
}
@@ -200,10 +202,11 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2099-01-02T01:21Z");
bundles[0].setProcessConcurrency(6);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
String startTime = TimeUtil.getTimeWrtSystemTime(1);
String endTime = TimeUtil.getTimeWrtSystemTime(40);
InstancesResult r = prism.getProcessHelper()
- .getProcessInstanceKill(processName, "?start=" + startTime + "&end=" + endTime);
+ .getProcessInstanceKill(processName, "?start=" + startTime + "&end=" + endTime);
LOGGER.info(r.getMessage());
Assert.assertEquals(r.getInstances(), null);
}
@@ -220,12 +223,14 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:21Z");
bundles[0].setProcessConcurrency(6);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
prism.getProcessHelper()
- .getProcessInstanceKill(processName, "?start=2010-01-02T01:05Z&end=2010-01-02T01:16Z");
- TimeUtil.sleepSeconds(TIMEOUT);
+ .getProcessInstanceKill(processName, "?start=2010-01-02T01:05Z&end=2010-01-02T01:16Z");
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
InstanceUtil.validateResponse(r, 5, 2, 0, 0, 3);
}
@@ -238,13 +243,15 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void testProcessInstanceKillLastInstance() throws Exception {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:21Z");
- bundles[0].setProcessConcurrency(6);
+ bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 10);
prism.getProcessHelper().getProcessInstanceKill(processName, "?start=2010-01-02T01:20Z");
- TimeUtil.sleepSeconds(TIMEOUT);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
InstanceUtil.validateResponse(r, 5, 4, 0, 0, 1);
}
@@ -259,9 +266,11 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:04Z");
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
prism.getProcessHelper().getProcessInstanceSuspend(processName, "?start=2010-01-02T01:00Z");
InstancesResult r = prism.getProcessHelper().getProcessInstanceKill(processName,
- "?start=2010-01-02T01:00Z");
+ "?start=2010-01-02T01:00Z");
InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.KILLED);
}
@@ -276,10 +285,12 @@ public class ProcessInstanceKillsTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:04Z");
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstanceUtil.waitTillInstanceReachState(serverOC.get(0), processName, 1,
- CoordinatorAction.Status.SUCCEEDED, EntityType.PROCESS);
+ CoordinatorAction.Status.SUCCEEDED, EntityType.PROCESS);
InstancesResult r = prism.getProcessHelper().getProcessInstanceKill(processName,
- "?start=2010-01-02T01:00Z");
+ "?start=2010-01-02T01:00Z");
InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.SUCCEEDED);
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRerunTest.java
----------------------------------------------------------------------
diff --git a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRerunTest.java b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRerunTest.java
index 4ae6d72..62a8b2b 100644
--- a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRerunTest.java
+++ b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRerunTest.java
@@ -23,12 +23,13 @@ import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.InstancesResult;
-import org.apache.falcon.regression.core.util.BundleUtil;
-import org.apache.falcon.regression.core.util.HadoopUtil;
import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.OozieUtil;
+import org.apache.falcon.regression.core.util.HadoopUtil;
+import org.apache.falcon.regression.core.util.BundleUtil;
import org.apache.falcon.regression.core.util.OSUtil;
-import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.core.util.Util;
+import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.Logger;
@@ -71,12 +72,7 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
HadoopUtil.uploadDir(clusterFS, aggregateWorkflowDir, OSUtil.RESOURCES_OOZIE);
Bundle b = BundleUtil.readELBundle();
b = new Bundle(b, cluster);
- String startDate = "2010-01-02T00:40Z";
- String endDate = "2010-01-02T01:20Z";
b.setInputFeedDataPath(feedInputPath);
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, endDate, 20);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.NORMAL_INPUT,
- b.getFeedDataPathPrefix(), dataDates);
}
@BeforeMethod(alwaysRun = true)
@@ -110,9 +106,10 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
bundles[0].setOutputFeedLocationData(feedOutputPath);
bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
- String process = bundles[0].getProcessData();
- TimeUtil.sleepSeconds(TIMEOUT);
- InstanceUtil.waitTillInstancesAreCreated(cluster, process, 0);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper().getProcessInstanceKill(processName,
start + "&end=2010-01-02T01:16Z");
InstanceUtil.validateResponse(r, 4, 0, 0, 0, 4);
@@ -135,7 +132,10 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
String process = bundles[0].getProcessData();
LOGGER.info("process: " + Util.prettyPrintXml(process));
bundles[0].submitFeedsScheduleProcess(prism);
- InstanceUtil.waitTillInstancesAreCreated(cluster, process, 0);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 3,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper()
.getProcessInstanceKill(processName, start + "&end=2010-01-02T01:11Z");
InstanceUtil.validateResponse(r, 3, 0, 0, 0, 3);
@@ -155,10 +155,12 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
public void testProcessInstanceRerunSomeKilled01() throws Exception {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:26Z");
bundles[0].setOutputFeedLocationData(feedOutputPath);
- bundles[0].setProcessConcurrency(5);
+ bundles[0].setProcessConcurrency(6);
bundles[0].submitFeedsScheduleProcess(prism);
- String process = bundles[0].getProcessData();
- InstanceUtil.waitTillInstancesAreCreated(cluster, process, 0);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 6,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper()
.getProcessInstanceKill(processName, start + "&end=2010-01-02T01:11Z");
InstanceUtil.validateResponse(r, 3, 0, 0, 0, 3);
@@ -180,8 +182,10 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
bundles[0].setOutputFeedLocationData(feedOutputPath);
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
- String process = bundles[0].getProcessData();
- InstanceUtil.waitTillInstancesAreCreated(cluster, process, 0);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 1,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
prism.getProcessHelper().getProcessInstanceKill(processName, start);
String wfID = InstanceUtil.getWorkflows(cluster, processName, Status.KILLED).get(0);
prism.getProcessHelper().getProcessInstanceRerun(processName, start);
@@ -202,8 +206,9 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
bundles[0].submitFeedsScheduleProcess(prism);
String process = bundles[0].getProcessData();
InstanceUtil.waitTillInstancesAreCreated(cluster, process, 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
String wfID = InstanceUtil.getWorkflows(cluster, processName, Status.RUNNING,
- Status.SUCCEEDED).get(0);
+ Status.SUCCEEDED).get(0);
InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 0, CoordinatorAction
.Status.SUCCEEDED, EntityType.PROCESS);
prism.getProcessHelper().getProcessInstanceRerun(processName, start);
@@ -222,12 +227,16 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
bundles[0].setOutputFeedLocationData(feedOutputPath);
bundles[0].setProcessConcurrency(2);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 2,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
prism.getProcessHelper().getProcessInstanceSuspend(processName,
start + "&end=2010-01-02T01:06Z");
prism.getProcessHelper().getProcessInstanceRerun(processName,
start + "&end=2010-01-02T01:06Z");
Assert.assertEquals(InstanceUtil.getInstanceStatus(cluster, processName, 0, 1),
- CoordinatorAction.Status.SUSPENDED);
+ CoordinatorAction.Status.SUSPENDED);
}
/**
@@ -241,8 +250,8 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
bundles[0].setOutputFeedLocationData(feedOutputPath);
bundles[0].setProcessConcurrency(3);
bundles[0].submitFeedsScheduleProcess(prism);
- String process = bundles[0].getProcessData();
- InstanceUtil.waitTillInstancesAreCreated(cluster, process, 0);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 2,
CoordinatorAction.Status.SUCCEEDED, EntityType.PROCESS);
List<String> wfIDs = InstanceUtil.getWorkflows(cluster, processName);
@@ -267,7 +276,7 @@ public class ProcessInstanceRerunTest extends BaseTestClass {
bundles[0].submitFeedsScheduleProcess(prism);
CoordinatorAction.Status s;
InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 1,
- CoordinatorAction.Status.TIMEDOUT, EntityType.PROCESS);
+ CoordinatorAction.Status.TIMEDOUT, EntityType.PROCESS);
prism.getProcessHelper().getProcessInstanceRerun(processName,
start + "&end=2010-01-02T01:11Z");
s = InstanceUtil.getInstanceStatus(cluster, processName, 0, 0);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceResumeTest.java
----------------------------------------------------------------------
diff --git a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceResumeTest.java b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceResumeTest.java
index a4c9619..f564bb4 100644
--- a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceResumeTest.java
+++ b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceResumeTest.java
@@ -18,21 +18,23 @@
package org.apache.falcon.regression;
+import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.regression.core.bundle.Bundle;
import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.InstancesResult;
-import org.apache.falcon.regression.core.response.InstancesResult.WorkflowStatus;
import org.apache.falcon.regression.core.response.ResponseKeys;
-import org.apache.falcon.regression.core.util.BundleUtil;
-import org.apache.falcon.regression.core.util.HadoopUtil;
import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.OozieUtil;
+import org.apache.falcon.regression.core.util.HadoopUtil;
+import org.apache.falcon.regression.core.util.BundleUtil;
import org.apache.falcon.regression.core.util.OSUtil;
-import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.core.util.Util;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.Logger;
+import org.apache.oozie.client.CoordinatorAction;
+import org.apache.oozie.client.OozieClient;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
@@ -41,7 +43,6 @@ import org.testng.annotations.Test;
import java.io.IOException;
import java.lang.reflect.Method;
-import java.util.List;
/**
* Process instance resume tests.
@@ -51,13 +52,12 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
private ColoHelper cluster = servers.get(0);
private FileSystem clusterFS = serverFS.get(0);
+ private OozieClient clusterOC = serverOC.get(0);
private String baseTestHDFSDir = baseHDFSDir + "/ProcessInstanceResumeTest";
private String feedInputPath = baseTestHDFSDir + "/input" + MINUTE_DATE_PATTERN;
private String feedOutputPath = baseTestHDFSDir + "/output-data" + MINUTE_DATE_PATTERN;
private String aggregateWorkflowDir = baseTestHDFSDir + "/aggregator";
private static final Logger LOGGER = Logger.getLogger(ProcessInstanceResumeTest.class);
- private static final double SCHEDULED = 15;
- private static final double AFFECTED = 10;
private String processName;
private String wholeRange = "?start=2010-01-02T01:00Z&end=2010-01-02T01:26Z";
@@ -67,12 +67,7 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
HadoopUtil.uploadDir(clusterFS, aggregateWorkflowDir, OSUtil.RESOURCES_OOZIE);
Bundle b = BundleUtil.readELBundle();
b = new Bundle(b, cluster);
- String startDate = "2010-01-01T23:20Z";
- String endDate = "2010-01-02T01:40Z";
b.setInputFeedDataPath(feedInputPath);
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, endDate, 20);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.NORMAL_INPUT,
- b.getFeedDataPathPrefix(), dataDates);
}
@BeforeMethod(alwaysRun = true)
@@ -99,23 +94,24 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
/**
* Schedule process. Suspend some instances. Attempt to -resume instance using single -end
- * parameter results in failure.
+ * parameter. Instances up to the end date (exclusive) will be resumed.
*
* @throws Exception
*/
@Test(groups = {"singleCluster"})
public void testProcessInstanceResumeOnlyEnd() throws Exception {
- bundles[0].setProcessPeriodicity(5, TimeUnit.minutes);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 6,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
prism.getProcessHelper().getProcessInstanceSuspend(processName,
"?start=2010-01-02T01:05Z&end=2010-01-02T01:21Z");
- TimeUtil.sleepSeconds(AFFECTED);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
wholeRange);
InstanceUtil.validateResponse(r, 6, 2, 4, 0, 0);
r = prism.getProcessHelper().getProcessInstanceResume(processName, "?end=2010-01-02T01:15Z");
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ InstanceUtil.validateResponse(r, 3, 3, 0, 0, 0);
}
/**
@@ -127,10 +123,12 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void testProcessInstanceResumeResumeSome() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 6,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
prism.getProcessHelper().getProcessInstanceSuspend(processName,
"?start=2010-01-02T01:05Z&end=2010-01-02T01:21Z");
- TimeUtil.sleepSeconds(AFFECTED);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
wholeRange);
InstanceUtil.validateResponse(r, 6, 2, 4, 0, 0);
@@ -149,10 +147,12 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void testProcessInstanceResumeResumeMany() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 6,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
String withinRange = "?start=2010-01-02T01:05Z&end=2010-01-02T01:21Z";
prism.getProcessHelper().getProcessInstanceSuspend(processName, withinRange);
- TimeUtil.sleepSeconds(AFFECTED);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
wholeRange);
InstanceUtil.validateResponse(r, 6, 2, 4, 0, 0);
@@ -171,14 +171,15 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
public void testProcessInstanceResumeSingle() throws Exception {
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 1,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 2);
String start = "?start=2010-01-02T01:00Z";
prism.getProcessHelper().getProcessInstanceSuspend(processName, start);
- TimeUtil.sleepSeconds(AFFECTED);
prism.getProcessHelper().getProcessInstanceResume(processName, start);
- TimeUtil.sleepSeconds(AFFECTED);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName, start);
- InstanceUtil.validateSuccessOnlyStart(r, WorkflowStatus.RUNNING);
+ InstanceUtil.validateResponse(r, 6, 1, 0, 5, 0);
}
/**
@@ -197,7 +198,7 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
/**
* Attempt to perform -resume action without time range parameters should fail with an
- + appropriate status code or message.
+ + appropriate status code or message. Will fail now due to jira: https://issues.apache.org/jira/browse/FALCON-710
*
* @throws Exception
*/
@@ -231,13 +232,15 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void testProcessInstanceResumeNonSuspended() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 6,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
String start = "?start=2010-01-02T01:05Z";
- prism.getProcessHelper().getProcessInstanceResume(processName, start);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName, start);
- InstanceUtil.validateResponse(r, 1, 1, 0, 0, 0);
+ InstanceUtil.validateResponse(r, 5, 5, 0, 0, 0);
r = prism.getProcessHelper().getProcessInstanceResume(processName, start);
- InstanceUtil.validateResponse(r, 1, 1, 0, 0, 0);
+ InstanceUtil.validateResponse(r, 5, 5, 0, 0, 0);
}
/**
@@ -250,40 +253,20 @@ public class ProcessInstanceResumeTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void testProcessInstanceResumeLastInstance() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 6,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
String last = "?start=2010-01-02T01:25Z";
prism.getProcessHelper().getProcessInstanceSuspend(processName, last);
- TimeUtil.sleepSeconds(AFFECTED);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
wholeRange);
InstanceUtil.validateResponse(r, 6, 5, 1, 0, 0);
- TimeUtil.sleepSeconds(10);
prism.getProcessHelper().getProcessInstanceResume(processName, last);
r = prism.getProcessHelper().getProcessInstanceStatus(processName, wholeRange);
InstanceUtil.validateResponse(r, 6, 6, 0, 0, 0);
}
- /**
- * Schedule process. Suspend all instances except the first and the last using appropriate
- * -start/-end parameters. Resume that instances. Check that there are no suspended ones.
- *
- * @throws Exception
- */
- @Test(groups = {"singleCluster"})
- public void testProcessInstanceResumeWithinRange() throws Exception {
- bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(SCHEDULED);
- String withinRange = "?start=2010-01-02T01:05Z&end=2010-01-02T01:21Z";
- prism.getProcessHelper().getProcessInstanceSuspend(processName, withinRange);
- TimeUtil.sleepSeconds(AFFECTED);
- InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- wholeRange);
- InstanceUtil.validateResponse(r, 6, 2, 4, 0, 0);
- prism.getProcessHelper().getProcessInstanceResume(processName, withinRange);
- r = prism.getProcessHelper().getProcessInstanceStatus(processName, wholeRange);
- InstanceUtil.validateResponse(r, 6, 6, 0, 0, 0);
- }
-
@AfterClass(alwaysRun = true)
public void tearDownClass() throws IOException {
cleanTestDirs();
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRunningTest.java
----------------------------------------------------------------------
diff --git a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRunningTest.java b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRunningTest.java
index 7a7e735..d3af1c0 100644
--- a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRunningTest.java
+++ b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceRunningTest.java
@@ -18,23 +18,27 @@
package org.apache.falcon.regression;
+import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.regression.core.bundle.Bundle;
import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.InstancesResult;
import org.apache.falcon.regression.core.response.InstancesResult.WorkflowStatus;
import org.apache.falcon.regression.core.response.ResponseKeys;
-import org.apache.falcon.regression.core.util.AssertUtil;
-import org.apache.falcon.regression.core.util.BundleUtil;
-import org.apache.falcon.regression.core.util.HadoopUtil;
import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.OozieUtil;
+import org.apache.falcon.regression.core.util.HadoopUtil;
+import org.apache.falcon.regression.core.util.BundleUtil;
import org.apache.falcon.regression.core.util.OSUtil;
-import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.core.util.Util;
+import org.apache.falcon.regression.core.util.TimeUtil;
+import org.apache.falcon.regression.core.util.AssertUtil;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.Logger;
+import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.Job;
+import org.apache.oozie.client.OozieClient;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -54,6 +58,7 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
private ColoHelper cluster = servers.get(0);
private FileSystem clusterFS = serverFS.get(0);
+ private OozieClient clusterOC = serverOC.get(0);
private String baseTestHDFSDir = baseHDFSDir + "/ProcessInstanceRunningTest";
private String aggregateWorkflowDir = baseTestHDFSDir + "/aggregator";
private String feedInputPath = baseTestHDFSDir + "/input" + MINUTE_DATE_PATTERN;
@@ -69,12 +74,7 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
Bundle bundle = BundleUtil.readELBundle();
bundle.generateUniqueBundle();
bundle = new Bundle(bundle, cluster);
- String startDate = "2010-01-02T00:40Z";
- String endDate = "2010-01-02T01:11Z";
bundle.setInputFeedDataPath(feedInputPath);
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, endDate, 20);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.NORMAL_INPUT,
- bundle.getFeedDataPathPrefix(), dataDates);
}
@BeforeMethod(alwaysRun = true)
@@ -107,7 +107,10 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
public void getResumedProcessInstance() throws Exception {
bundles[0].setProcessConcurrency(3);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 3,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
String process = bundles[0].getProcessData();
AssertUtil.assertSucceeded(prism.getProcessHelper().suspend(process));
TimeUtil.sleepSeconds(TIMEOUT);
@@ -127,6 +130,10 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
public void getSuspendedProcessInstance() throws Exception {
bundles[0].setProcessConcurrency(3);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 3,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
AssertUtil.assertSucceeded(prism.getProcessHelper().suspend(bundles[0].getProcessData()));
TimeUtil.sleepSeconds(TIMEOUT);
InstancesResult r = prism.getProcessHelper().getRunningInstance(processName);
@@ -142,6 +149,8 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void getRunningProcessInstance() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstancesResult r = prism.getProcessHelper().getRunningInstance(processName);
InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.RUNNING);
}
@@ -155,7 +164,7 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
public void getNonExistenceProcessInstance() throws Exception {
InstancesResult r = prism.getProcessHelper().getRunningInstance("invalidName");
Assert.assertEquals(r.getStatusCode(), ResponseKeys.PROCESS_NOT_FOUND,
- "Unexpected status code");
+ "Unexpected status code");
}
/**
@@ -169,7 +178,7 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
prism.getProcessHelper().delete(bundles[0].getProcessData());
InstancesResult r = prism.getProcessHelper().getRunningInstance(processName);
Assert.assertEquals(r.getStatusCode(), ResponseKeys.PROCESS_NOT_FOUND,
- "Unexpected status code");
+ "Unexpected status code");
}
/**
@@ -181,6 +190,8 @@ public class ProcessInstanceRunningTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void getSucceededProcessInstance() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstanceUtil.waitForBundleToReachState(cluster, processName, Job.Status.SUCCEEDED);
InstancesResult r = prism.getProcessHelper().getRunningInstance(processName);
InstanceUtil.validateSuccessWOInstances(r);
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceStatusTest.java
----------------------------------------------------------------------
diff --git a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceStatusTest.java b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceStatusTest.java
index ad731bf..736ed0c 100644
--- a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceStatusTest.java
+++ b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceStatusTest.java
@@ -25,20 +25,23 @@ import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.InstancesResult;
import org.apache.falcon.regression.core.response.InstancesResult.WorkflowStatus;
import org.apache.falcon.regression.core.response.ResponseKeys;
-import org.apache.falcon.regression.core.util.AssertUtil;
-import org.apache.falcon.regression.core.util.BundleUtil;
-import org.apache.falcon.regression.core.util.HadoopUtil;
import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.OozieUtil;
import org.apache.falcon.regression.core.util.OSUtil;
-import org.apache.falcon.regression.core.util.TimeUtil;
+import org.apache.falcon.regression.core.util.BundleUtil;
import org.apache.falcon.regression.core.util.Util;
+import org.apache.falcon.regression.core.util.TimeUtil;
+import org.apache.falcon.regression.core.util.HadoopUtil;
+import org.apache.falcon.regression.core.util.AssertUtil;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.log4j.Logger;
+import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.CoordinatorAction.Status;
import org.apache.oozie.client.Job;
import org.apache.oozie.client.OozieClient;
+import org.apache.oozie.client.OozieClientException;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
@@ -50,7 +53,6 @@ import javax.xml.bind.JAXBException;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.URISyntaxException;
-import java.util.List;
/**
* Process instance status tests.
@@ -80,13 +82,7 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
Bundle bundle = BundleUtil.readELBundle();
bundle.generateUniqueBundle();
bundle = new Bundle(bundle, cluster);
- String startDate = "2010-01-01T23:40Z";
- String endDate = "2010-01-02T02:40Z";
bundle.setInputFeedDataPath(feedInputPath);
- String prefix = bundle.getFeedDataPathPrefix();
- HadoopUtil.deleteDirIfExists(prefix.substring(1), clusterFS);
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, endDate, 20);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.NORMAL_INPUT, prefix, dataDates);
}
/**
@@ -124,6 +120,7 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
bundles[0].setProcessPeriodicity(1, TimeUnit.minutes);
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
"?start=2010-01-02T01:00Z&end=2010-01-02T10:20Z");
InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.RUNNING);
@@ -151,7 +148,7 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
/**
* Schedule process. Perform -getStatus using -end parameter which is out of process
- * validity range. Attempt should fail.
+ * validity range. Attempt should succeed with end defaulted to entity end.
*
* @throws Exception
*/
@@ -161,20 +158,25 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
bundles[0].submitFeedsScheduleProcess(prism);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
"?start=2010-01-02T01:00Z&end=2010-01-02T01:30Z");
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ InstanceUtil.validateResponse(r, 5, 0, 0, 5, 0);
}
/**
- * Schedule process and try to -getStatus without date parameters. Attempt should fail with
- * an appropriate message.
+ * Schedule process and try to -getStatus without date parameters. Attempt should succeed. Start defaults
+ * to start of entity and end defaults to end of entity.
*/
@Test(groups = {"singleCluster"})
public void testProcessInstanceStatusDateEmpty()
- throws JAXBException, AuthenticationException, IOException, URISyntaxException {
- bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T02:30Z");
+ throws JAXBException, AuthenticationException, IOException, URISyntaxException, OozieClientException {
+ bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:30Z");
+ bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(serverOC.get(0), processName, 5 ,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName, null);
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ InstanceUtil.validateResponse(r, 6, 5, 0, 1, 0);
}
/**
@@ -187,23 +189,28 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
public void testProcessInstanceStatusStartAndEnd() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:20Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:20Z");
InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.RUNNING);
}
/**
* Schedule process. Perform -getStatus using -start parameter which is out of process
- * validity range. Attempt should fail.
+ * validity range. Attempt should succeed, with start defaulted to entity start time.
*
* @throws Exception
*/
@Test(groups = {"singleCluster"})
public void testProcessInstanceStatusStartOutOfRange() throws Exception {
bundles[0].setOutputFeedPeriodicity(5, TimeUnit.minutes);
+ bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T00:00Z&end=2010-01-02T01:20Z");
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ "?start=2010-01-02T00:00Z&end=2010-01-02T01:21Z");
+ InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.RUNNING);
}
/**
@@ -217,7 +224,7 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
bundles[0].submitFeedsScheduleProcess(prism);
AssertUtil.assertSucceeded(prism.getProcessHelper().delete(bundles[0].getProcessData()));
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:20Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:20Z");
if ((r.getStatusCode() != ResponseKeys.PROCESS_NOT_FOUND)) {
Assert.assertTrue(false);
}
@@ -235,13 +242,13 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
AssertUtil.assertSucceeded(prism.getProcessHelper().suspend(bundles[0].getProcessData()));
TimeUtil.sleepSeconds(TIMEOUT);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z");
+ "?start=2010-01-02T01:00Z");
InstanceUtil.validateSuccessOnlyStart(r, WorkflowStatus.SUSPENDED);
}
/**
* Schedule process. Try to -getStatus using -start/-end parameters with values which were
- * reversed i.e. -start is further then -end. Attempt should fail.
+ * reversed i.e. -start is further than -end. Attempt should fail.
*
* @throws Exception
*/
@@ -249,13 +256,13 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
public void testProcessInstanceStatusReverseDateRange() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:20Z&end=2010-01-02T01:07Z");
+ "?start=2010-01-02T01:20Z&end=2010-01-02T01:07Z");
InstanceUtil.validateSuccessWithStatusCode(r, 400);
}
/**
* Schedule process. Perform -getStatus using -start/-end parameters which are out of process
- * validity range. Attempt should fail.
+ * validity range. Attempt should succeed, with start/end defaulted to entity's start/end.
*
* @throws Exception
*/
@@ -265,11 +272,13 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
bundles[0].setOutputFeedLocationData(feedOutputPath);
bundles[0].setProcessConcurrency(2);
bundles[0].submitFeedsScheduleProcess(prism);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 2,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
"?start=2010-01-02T00:00Z&end=2010-01-02T01:30Z");
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ InstanceUtil.validateResponse(r, 5, 2, 0, 3, 0);
}
/**
@@ -282,10 +291,13 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
public void testProcessInstanceStatusResumed() throws Exception {
bundles[0].setOutputFeedPeriodicity(5, TimeUnit.minutes);
bundles[0].setOutputFeedLocationData(feedOutputPath);
- bundles[0].setProcessConcurrency(2);
+ bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
String process = bundles[0].getProcessData();
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, process, Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
prism.getProcessHelper().suspend(process);
AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, process, Job.Status.SUSPENDED);
prism.getProcessHelper().resume(process);
@@ -306,11 +318,13 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
@Test(groups = {"singleCluster"})
public void testProcessInstanceStatusOnlyStart() throws Exception {
bundles[0].submitFeedsScheduleProcess(prism);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 1,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
"?start=2010-01-02T01:00Z");
- InstanceUtil.validateSuccessOnlyStart(r, WorkflowStatus.RUNNING);
+ InstanceUtil.validateResponse(r, 5, 1, 0, 4, 0);
}
/**
@@ -331,40 +345,22 @@ public class ProcessInstanceStatusTest extends BaseTestClass {
}
/**
- * Schedule process. Suspend it. -getStatus of it's instances. Check if response reflects
- * their status as suspended.
+ * Schedule process. Try to -getStatus without time range parameters. Attempt succeeds.
*
- * @throws Exception
- */
- @Test(groups = {"singleCluster"})
- public void testProcessInstanceStatusSuspended() throws Exception {
- for (int i = 0; i < bundles[0].getClusters().size(); i++) {
- LOGGER.info("cluster to be submitted: " + i + " "
- + Util.prettyPrintXml(bundles[0].getClusters().get(i)));
- }
- String process = bundles[0].getProcessData();
- bundles[0].submitFeedsScheduleProcess(prism);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, process, Job.Status.RUNNING);
- AssertUtil.assertSucceeded(prism.getProcessHelper().suspend(process));
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, process, Job.Status.SUSPENDED);
- TimeUtil.sleepSeconds(TIMEOUT);
- InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:20Z");
- InstanceUtil.validateSuccess(r, bundles[0], WorkflowStatus.SUSPENDED);
- }
-
- /**
- * Schedule process. Try to -getStatus without time range parameters. Attempt should fails
- * with an appropriate status code.
*
* @throws Exception
*/
@Test(groups = {"singleCluster"})
public void testProcessInstanceStatusWoParams() throws Exception {
- bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T02:30Z");
+ bundles[0].setProcessConcurrency(5);
+ bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:23Z");
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper().getProcessInstanceStatus(processName, null);
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ InstanceUtil.validateResponse(r, 5, 5, 0, 0, 0);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/94eb546b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceSuspendTest.java
----------------------------------------------------------------------
diff --git a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceSuspendTest.java b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceSuspendTest.java
index 9497a49..bcdd704 100644
--- a/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceSuspendTest.java
+++ b/falcon-regression/merlin/src/test/java/org/apache/falcon/regression/ProcessInstanceSuspendTest.java
@@ -23,15 +23,14 @@ import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.InstancesResult;
-import org.apache.falcon.regression.core.response.InstancesResult.WorkflowStatus;
import org.apache.falcon.regression.core.response.ResponseKeys;
-import org.apache.falcon.regression.core.util.AssertUtil;
-import org.apache.falcon.regression.core.util.BundleUtil;
-import org.apache.falcon.regression.core.util.HadoopUtil;
import org.apache.falcon.regression.core.util.InstanceUtil;
+import org.apache.falcon.regression.core.util.HadoopUtil;
+import org.apache.falcon.regression.core.util.BundleUtil;
+import org.apache.falcon.regression.core.util.OozieUtil;
import org.apache.falcon.regression.core.util.OSUtil;
-import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.core.util.Util;
+import org.apache.falcon.regression.core.util.AssertUtil;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.Logger;
@@ -62,7 +61,6 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
private ColoHelper cluster = servers.get(0);
private FileSystem clusterFS = serverFS.get(0);
private static final Logger LOGGER = Logger.getLogger(ProcessInstanceSuspendTest.class);
- private static final double TIMEOUT = 15;
private String processName;
private OozieClient clusterOC = serverOC.get(0);
@@ -70,16 +68,9 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
public void createTestData() throws Exception {
LOGGER.info("in @BeforeClass");
HadoopUtil.uploadDir(clusterFS, aggregateWorkflowDir, OSUtil.RESOURCES_OOZIE);
-
Bundle bundle = BundleUtil.readELBundle();
bundle = new Bundle(bundle, cluster);
- String startDate = "2010-01-01T23:40Z";
- String endDate = "2010-01-02T01:40Z";
bundle.setInputFeedDataPath(feedInputPath);
-
- List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, endDate, 20);
- HadoopUtil.flattenAndPutDataInFolder(clusterFS, OSUtil.NORMAL_INPUT,
- bundle.getFeedDataPathPrefix(), dataDates);
}
@BeforeMethod(alwaysRun = true)
@@ -102,7 +93,7 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
/**
* Schedule process. Try to suspend instances with start/end parameters which are
- * wider then process validity range. Should fail.
+ * wider then process validity range. Succeeds.
*
* @throws Exception
*/
@@ -111,17 +102,18 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:23Z");
bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult result = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
InstanceUtil.validateResponse(result, 5, 5, 0, 0, 0);
prism.getProcessHelper().getProcessInstanceSuspend(processName,
- "?start=2010-01-02T00:00Z&end=2010-01-02T01:30Z");
+ "?start=2010-01-02T00:00Z&end=2010-01-02T01:30Z");
result = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T00:00Z&end=2010-01-02T01:30Z");
- InstanceUtil.validateSuccessWithStatusCode(result, 0);
+ "?start=2010-01-02T00:00Z&end=2010-01-02T01:30Z");
+ InstanceUtil.validateResponse(result, 5, 0, 5, 0, 0);
}
/**
@@ -135,10 +127,12 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:04Z");
bundles[0].setProcessConcurrency(1);
bundles[0].submitFeedsScheduleProcess(prism);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
InstanceUtil.waitTillInstanceReachState(clusterOC, Util.getProcessName(bundles[0]
.getProcessData()), 1, CoordinatorAction.Status.SUCCEEDED, EntityType.PROCESS);
InstancesResult r = prism.getProcessHelper().getProcessInstanceSuspend(processName,
- "?start=2010-01-02T01:00Z");
+ "?start=2010-01-02T01:00Z");
InstanceUtil.validateSuccessWithStatusCode(r, 0);
}
@@ -153,34 +147,32 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:23Z");
bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult result = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
InstanceUtil.validateResponse(result, 5, 5, 0, 0, 0);
prism.getProcessHelper().getProcessInstanceSuspend(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
result = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
InstanceUtil.validateResponse(result, 5, 0, 5, 0, 0);
}
/**
* Schedule process and try to perform -suspend action without date range parameters.
- * Attempt should fail.
+ * Attempt should fail. Will fail because of jira : https://issues.apache.org/jira/browse/FALCON-710
*
* @throws Exception
*/
@Test(groups = {"singleCluster"})
public void testProcessInstanceSuspendWoParams() throws Exception {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:22Z");
- bundles[0].setProcessConcurrency(2);
bundles[0].submitFeedsScheduleProcess(prism);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
InstancesResult r = prism.getProcessHelper().getProcessInstanceSuspend(processName, null);
- InstanceUtil.validateSuccessWithStatusCode(r, 0);
+ InstanceUtil.validateSuccessWithStatusCode(r, ResponseKeys.UNPARSEABLE_DATE);
}
/**
@@ -194,16 +186,17 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:23Z");
bundles[0].setProcessConcurrency(3);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 3,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult result = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:22Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:22Z");
InstanceUtil.validateResponse(result, 5, 3, 0, 2, 0);
prism.getProcessHelper().getProcessInstanceSuspend(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:15Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:15Z");
result = prism.getProcessHelper().getProcessInstanceStatus(processName,
- "?start=2010-01-02T01:00Z&end=2010-01-02T01:22Z");
+ "?start=2010-01-02T01:00Z&end=2010-01-02T01:22Z");
InstanceUtil.validateResponse(result, 5, 0, 3, 2, 0);
}
@@ -217,10 +210,8 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:23Z");
bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
InstancesResult r = prism.getProcessHelper()
- .getProcessInstanceSuspend("invalidName", "?start=2010-01-02T01:20Z");
+ .getProcessInstanceSuspend("invalidName", "?start=2010-01-02T01:20Z");
if ((r.getStatusCode() != ResponseKeys.PROCESS_NOT_FOUND)) {
Assert.assertTrue(false);
}
@@ -228,7 +219,7 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
/**
* Schedule process. Perform -suspend action using only -start parameter which points to start
- * time of process. Check that only 1 instance is suspended then.
+ * time of process. Attempt suspends all instances
*
* @throws Exception
*/
@@ -237,14 +228,12 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:11Z");
bundles[0].setProcessConcurrency(3);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
- prism.getProcessHelper().getRunningInstance(processName);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 3,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult r = prism.getProcessHelper().getProcessInstanceSuspend(processName,
- "?start=2010-01-02T01:00Z");
- InstanceUtil.validateSuccessOnlyStart(r, WorkflowStatus.SUSPENDED);
- prism.getProcessHelper().getRunningInstance(processName);
+ "?start=2010-01-02T01:00Z");
+ InstanceUtil.validateResponse(r, 3, 0, 3, 0, 0);
}
/**
@@ -259,9 +248,10 @@ public class ProcessInstanceSuspendTest extends BaseTestClass {
bundles[0].setProcessValidity("2010-01-02T01:00Z", "2010-01-02T01:23Z");
bundles[0].setProcessConcurrency(5);
bundles[0].submitFeedsScheduleProcess(prism);
- TimeUtil.sleepSeconds(TIMEOUT);
- AssertUtil.checkStatus(clusterOC, EntityType.PROCESS, bundles[0].getProcessData(),
- Job.Status.RUNNING);
+ InstanceUtil.waitTillInstancesAreCreated(cluster, bundles[0].getProcessData(), 0);
+ OozieUtil.createMissingDependencies(cluster, EntityType.PROCESS, processName, 0);
+ InstanceUtil.waitTillInstanceReachState(clusterOC, processName, 5,
+ CoordinatorAction.Status.RUNNING, EntityType.PROCESS, 5);
InstancesResult result = prism.getProcessHelper().getProcessInstanceStatus(processName,
"?start=2010-01-02T01:00Z&end=2010-01-02T01:21Z");
InstanceUtil.validateResponse(result, 5, 5, 0, 0, 0);