You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by hu...@apache.org on 2019/11/26 17:59:19 UTC
[helix] branch master updated: Stabilize 5 unstable tests (#624)
This is an automated email from the ASF dual-hosted git repository.
hulee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
The following commit(s) were added to refs/heads/master by this push:
new 9280f22 Stabilize 5 unstable tests (#624)
9280f22 is described below
commit 9280f2201db9572b2582270e8e5663c2520b15c1
Author: Ali Reza Zamani Zadeh Najari <an...@linkedin.com>
AuthorDate: Tue Nov 26 09:59:09 2019 -0800
Stabilize 5 unstable tests (#624)
In this commit three test that were unstable for some time has been stabilized.
1- TestWorkflowControllerDataProvider:
This test were rely on thread sleep. This has been changed to TestHelper.verify.
2-TestDrop:
This test were rely on thread sleep. This has been changed to TestHelper.verify.
3-TestGetLastScheduledTaskExecInfo:
This test were rely on thread sleep. This has been changed to TestHelper.verify.
Also a check has been added to measure the scheduled tasks and compared the
result with the expected number of tasks that should be scheduled.
Also in this test some of the hard coded strings for queue namning
has been removed and replaced with TestHelper.getTestMethodName().
4- TestStopWorkflow:
At the beggining of each test, participant have started and at the end of the
test, participants have stopped. Hence dependency between the tests added to
avoid potential conflicts between individual tests. Also one global varible
has been use among the task to stop which has been changed to local variable.
Also Thread.sleep has been eliminated.
5- TestClusterVerifier:
In this test, Thread.sleep has been removed.
---
.../TestWorkflowControllerDataProvider.java | 35 +++++---
.../org/apache/helix/integration/TestDrop.java | 16 ++--
.../helix/integration/task/TestStopWorkflow.java | 50 ++++++++---
.../task/TestGetLastScheduledTaskExecInfo.java | 98 +++++++++++++++-------
.../apache/helix/tools/TestClusterVerifier.java | 21 +++--
5 files changed, 157 insertions(+), 63 deletions(-)
diff --git a/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestWorkflowControllerDataProvider.java b/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestWorkflowControllerDataProvider.java
index b4bf2dc..e4f05fd 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestWorkflowControllerDataProvider.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/dataproviders/TestWorkflowControllerDataProvider.java
@@ -30,22 +30,26 @@ import org.testng.annotations.Test;
public class TestWorkflowControllerDataProvider extends TaskTestBase {
@Test
- public void testResourceConfigRefresh() throws InterruptedException {
+ public void testResourceConfigRefresh() throws Exception {
Workflow.Builder builder = new Workflow.Builder("TEST");
JobConfig.Builder jobBuilder = JobConfig.Builder.fromMap(WorkflowGenerator.DEFAULT_JOB_CONFIG);
builder.addJob(WorkflowGenerator.JOB_NAME_1, jobBuilder);
_driver.start(builder.build());
- Thread.sleep(4000);
+
WorkflowControllerDataProvider cache =
new WorkflowControllerDataProvider("CLUSTER_" + TestHelper.getTestClassName());
- cache.requireFullRefresh();
- cache.refresh(_manager.getHelixDataAccessor());
- Assert.assertEquals(cache.getJobConfigMap().size(), 1);
- Assert.assertEquals(cache.getWorkflowConfigMap().size(), 1);
- Assert.assertEquals(cache.getContexts().size(), 2);
+ boolean expectedValuesAchieved = TestHelper.verify(() -> {
+ cache.requireFullRefresh();
+ cache.refresh(_manager.getHelixDataAccessor());
+ int configMapSize = cache.getJobConfigMap().size();
+ int workflowConfigMapSize = cache.getWorkflowConfigMap().size();
+ int contextsSize = cache.getContexts().size();
+ return (configMapSize == 1 && workflowConfigMapSize == 1 && contextsSize == 2);
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(expectedValuesAchieved);
builder = new Workflow.Builder("TEST1");
builder.addParentChildDependency(WorkflowGenerator.JOB_NAME_1, WorkflowGenerator.JOB_NAME_2);
@@ -54,11 +58,16 @@ public class TestWorkflowControllerDataProvider extends TaskTestBase {
builder.addJob(WorkflowGenerator.JOB_NAME_2, jobBuilder);
_driver.start(builder.build());
- Thread.sleep(4000);
- cache.requireFullRefresh();
- cache.refresh(_manager.getHelixDataAccessor());
- Assert.assertEquals(cache.getJobConfigMap().size(), 3);
- Assert.assertEquals(cache.getWorkflowConfigMap().size(), 2);
- Assert.assertEquals(cache.getContexts().size(), 5);
+
+ expectedValuesAchieved = TestHelper.verify(() -> {
+ cache.requireFullRefresh();
+ cache.refresh(_manager.getHelixDataAccessor());
+ int configMapSize = cache.getJobConfigMap().size();
+ int workflowConfigMapSize = cache.getWorkflowConfigMap().size();
+ int contextsSize = cache.getContexts().size();
+ return (configMapSize == 3 && workflowConfigMapSize == 2 && contextsSize == 5);
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(expectedValuesAchieved);
+
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestDrop.java b/helix-core/src/test/java/org/apache/helix/integration/TestDrop.java
index 4dcc739..d0a3d1d 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/TestDrop.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestDrop.java
@@ -56,16 +56,24 @@ public class TestDrop extends ZkTestBase {
* @param participants
*/
private void assertEmptyCSandEV(String clusterName, String db,
- MockParticipantManager[] participants) {
+ MockParticipantManager[] participants) throws Exception {
HelixDataAccessor accessor =
new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<>(_gZkClient));
PropertyKey.Builder keyBuilder = accessor.keyBuilder();
- Assert.assertNull(accessor.getProperty(keyBuilder.externalView(db)));
+ boolean isExternalViewNull = TestHelper.verify(() -> {
+ ExternalView externalView = accessor.getProperty(keyBuilder.externalView(db));
+ return (externalView == null);
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(isExternalViewNull);
for (MockParticipantManager participant : participants) {
String instanceName = participant.getInstanceName();
String sessionId = participant.getSessionId();
- Assert.assertNull(accessor.getProperty(keyBuilder.currentState(instanceName, sessionId, db)));
+ boolean isCurrentStateNull = TestHelper.verify(() -> {
+ CurrentState currentState = accessor.getProperty(keyBuilder.currentState(instanceName, sessionId, db));
+ return (currentState == null);
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(isCurrentStateNull);
}
}
@@ -109,8 +117,6 @@ public class TestDrop extends ZkTestBase {
HelixAdmin admin = new ZKHelixAdmin(_gZkClient);
admin.dropResource(clusterName, "TestDB0");
- Thread.sleep(1000);
-
assertEmptyCSandEV(clusterName, "TestDB0", participants);
controller.syncStop();
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestStopWorkflow.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestStopWorkflow.java
index 0721b88..84ce042 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestStopWorkflow.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestStopWorkflow.java
@@ -29,6 +29,7 @@ import org.apache.helix.TestHelper;
import org.apache.helix.integration.manager.MockParticipantManager;
import org.apache.helix.participant.StateMachineEngine;
import org.apache.helix.task.JobConfig;
+import org.apache.helix.task.JobContext;
import org.apache.helix.task.JobQueue;
import org.apache.helix.task.TaskCallbackContext;
import org.apache.helix.task.TaskConfig;
@@ -45,7 +46,6 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
public class TestStopWorkflow extends TaskTestBase {
- private boolean _taskFinishFlag = false;
@BeforeClass
public void beforeClass() throws Exception {
@@ -88,8 +88,8 @@ public class TestStopWorkflow extends TaskTestBase {
* Tests that stopping a workflow does result in its task ending up in STOPPED state.
* @throws InterruptedException
*/
- @Test
- public void testStopTask() throws InterruptedException {
+ @Test(dependsOnMethods = "testStopWorkflow")
+ public void testStopTask() throws Exception {
stopTestSetup(1);
String workflowName = TestHelper.getTestMethodName();
@@ -123,8 +123,8 @@ public class TestStopWorkflow extends TaskTestBase {
* Tests that stop() indeed frees up quotas for tasks belonging to the stopped workflow.
* @throws InterruptedException
*/
- @Test
- public void testStopTaskForQuota() throws InterruptedException {
+ @Test(dependsOnMethods = "testStopTask")
+ public void testStopTaskForQuota() throws Exception {
stopTestSetup(1);
String workflowNameToStop = TestHelper.getTestMethodName();
@@ -183,8 +183,8 @@ public class TestStopWorkflow extends TaskTestBase {
* Test that there is no thread leak when stopping and resuming.
* @throws InterruptedException
*/
- @Test
- public void testResumeTaskForQuota() throws InterruptedException {
+ @Test(dependsOnMethods = "testStopTaskForQuota")
+ public void testResumeTaskForQuota() throws Exception {
stopTestSetup(1);
String workflowName_1 = TestHelper.getTestMethodName();
@@ -204,13 +204,36 @@ public class TestStopWorkflow extends TaskTestBase {
_driver.start(workflowBuilder_1.build());
- Thread.sleep(2000L); // Sleep until each task really is in progress
+ // Check the jobs are in progress and the tasks are running.
+ // Each job has one task. Hence, we just check the state of the partition 0.
+ for (int i = 0; i < 30; i++) {
+ String jobName = workflowName_1 + "_JOB" + i;
+ _driver.pollForJobState(workflowName_1, jobName, TaskState.IN_PROGRESS);
+ boolean isTaskInRunningState = TestHelper.verify(() -> {
+ JobContext jobContext = _driver.getJobContext(jobName);
+ String state = jobContext.getMapField(0).get("STATE");
+ return (state!= null && state.equals("RUNNING"));
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(isTaskInRunningState);
+ }
+
_driver.stop(workflowName_1);
_driver.pollForWorkflowState(workflowName_1, TaskState.STOPPED);
- _taskFinishFlag = false;
_driver.resume(workflowName_1);
- Thread.sleep(2000L); // Sleep until each task really is in progress
+
+ // Check the jobs are in progress and the tasks are running.
+ // Each job has one task. Hence, we just check the state of the partition 0.
+ for (int i = 0; i < 30; i++) {
+ String jobName = workflowName_1 + "_JOB" + i;
+ _driver.pollForJobState(workflowName_1, jobName, TaskState.IN_PROGRESS);
+ boolean isTaskInRunningState = TestHelper.verify(() -> {
+ JobContext jobContext = _driver.getJobContext(jobName);
+ String state = jobContext.getMapField(0).get("STATE");
+ return (state!= null && state.equals("RUNNING"));
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(isTaskInRunningState);
+ }
// By now there should only be 30 threads occupied
@@ -276,13 +299,16 @@ public class TestStopWorkflow extends TaskTestBase {
* A mock task class that models a short-lived task to be stopped.
*/
private class StopTask extends MockTask {
+ private boolean _stopFlag = false;
+
StopTask(TaskCallbackContext context) {
super(context);
}
@Override
public TaskResult run() {
- while (!_taskFinishFlag) {
+ _stopFlag = false;
+ while (!_stopFlag) {
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
@@ -302,7 +328,7 @@ public class TestStopWorkflow extends TaskTestBase {
@Override
public void cancel() {
- _taskFinishFlag = true;
+ _stopFlag = true;
}
}
}
diff --git a/helix-core/src/test/java/org/apache/helix/task/TestGetLastScheduledTaskExecInfo.java b/helix-core/src/test/java/org/apache/helix/task/TestGetLastScheduledTaskExecInfo.java
index 4be4545..1697183 100644
--- a/helix-core/src/test/java/org/apache/helix/task/TestGetLastScheduledTaskExecInfo.java
+++ b/helix-core/src/test/java/org/apache/helix/task/TestGetLastScheduledTaskExecInfo.java
@@ -30,10 +30,14 @@ import org.apache.helix.integration.task.TaskTestUtil;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import org.apache.helix.TestHelper;
public class TestGetLastScheduledTaskExecInfo extends TaskTestBase {
private final static String TASK_START_TIME_KEY = "START_TIME";
private final static long INVALID_TIMESTAMP = -1L;
+ private static final long SHORT_EXECUTION_TIME = 10L;
+ private static final long LONG_EXECUTION_TIME = 99999999L;
+ private static final long DELETE_DELAY = 30000L;
@BeforeClass
public void beforeClass() throws Exception {
@@ -42,37 +46,56 @@ public class TestGetLastScheduledTaskExecInfo extends TaskTestBase {
}
@Test
- public void testGetLastScheduledTaskExecInfo() throws InterruptedException {
- List<Long> startTimesWithStuckTasks = setupTasks("TestWorkflow_2", 5, 99999999);
+ public void testGetLastScheduledTaskExecInfo() throws Exception {
+ // Start new queue that has one job with long tasks and record start time of the tasks
+ String queueName = TestHelper.getTestMethodName();
+ // Create and start new queue that has one job with 5 tasks.
+ // Each task has a long execution time.
+ // Since NumConcurrentTasksPerInstance is equal to 2, here we wait until two tasks have
+ // been scheduled (expectedScheduledTime = 2).
+ List<Long> startTimesWithStuckTasks = setupTasks(queueName, 5, LONG_EXECUTION_TIME, 2);
// Wait till the job is in progress
- _driver.pollForJobState("TestWorkflow_2", "TestWorkflow_2_job_0", TaskState.IN_PROGRESS);
+ _driver.pollForJobState(queueName, queueName + "_job_0", TaskState.IN_PROGRESS);
// First two must be -1 (two tasks are stuck), and API call must return the last value (most
// recent timestamp)
Assert.assertEquals(startTimesWithStuckTasks.get(0).longValue(), INVALID_TIMESTAMP);
Assert.assertEquals(startTimesWithStuckTasks.get(1).longValue(), INVALID_TIMESTAMP);
- Long lastScheduledTaskTs = _driver.getLastScheduledTaskTimestamp("TestWorkflow_2");
- // Get the last timestamp (which will be most recent)
- Assert.assertEquals(startTimesWithStuckTasks.get(4), lastScheduledTaskTs);
- TaskExecutionInfo execInfo = _driver.getLastScheduledTaskExecutionInfo("TestWorkflow_2");
- Assert.assertEquals(execInfo.getJobName(), "TestWorkflow_2_job_0");
- // Workflow 2 will be stuck, so its partition state will be RUNNING
- Assert.assertEquals(execInfo.getTaskPartitionState(), TaskPartitionState.RUNNING);
- Assert.assertEquals(execInfo.getStartTimeStamp(), lastScheduledTaskTs);
+ // Workflow will be stuck so its partition state will be Running
+ boolean hasQueueReachedDesiredState = TestHelper.verify(() -> {
+ Long lastScheduledTaskTs = _driver.getLastScheduledTaskTimestamp(queueName);
+ TaskExecutionInfo execInfo = _driver.getLastScheduledTaskExecutionInfo(queueName);
+ return (execInfo.getJobName().equals(queueName + "_job_0")
+ && execInfo.getTaskPartitionState() == TaskPartitionState.RUNNING
+ && execInfo.getStartTimeStamp().equals(lastScheduledTaskTs)
+ && startTimesWithStuckTasks.get(4).equals(lastScheduledTaskTs));
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(hasQueueReachedDesiredState);
- List<Long> startTimesFastTasks = setupTasks("TestWorkflow_3", 4, 10);
- // Wait till the job is in progress
- _driver.pollForJobState("TestWorkflow_3", "TestWorkflow_3_job_0", TaskState.IN_PROGRESS);
- // API call needs to return the most recent timestamp (value at last index)
- lastScheduledTaskTs = _driver.getLastScheduledTaskTimestamp("TestWorkflow_3");
- execInfo = _driver.getLastScheduledTaskExecutionInfo("TestWorkflow_3");
-
- Assert.assertEquals(startTimesFastTasks.get(startTimesFastTasks.size() - 1),
- lastScheduledTaskTs);
- Assert.assertEquals(execInfo.getJobName(), "TestWorkflow_3_job_0");
- Assert.assertEquals(execInfo.getTaskPartitionState(), TaskPartitionState.COMPLETED);
- Assert.assertEquals(execInfo.getStartTimeStamp(), lastScheduledTaskTs);
+ // Stop and delete the queue
+ _driver.stop(queueName);
+ _driver.deleteAndWaitForCompletion(queueName, DELETE_DELAY);
+
+ // Start the new queue with new task configuration.
+ // Create and start new queue that has one job with 4 tasks.
+ // Each task has a short execution time. In the setupTasks we wait until all of the tasks have
+ // been scheduled (expectedScheduledTime = 4).
+ List<Long> startTimesFastTasks = setupTasks(queueName, 4, SHORT_EXECUTION_TIME, 4);
+ // Wait till the job is in progress or completed. Since the tasks have short execution time, we
+ // wait for either IN_PROGRESS or COMPLETED states
+ _driver.pollForJobState(queueName, queueName + "_job_0", TaskState.IN_PROGRESS,
+ TaskState.COMPLETED);
+
+ hasQueueReachedDesiredState = TestHelper.verify(() -> {
+ Long lastScheduledTaskTs = _driver.getLastScheduledTaskTimestamp(queueName);
+ TaskExecutionInfo execInfo = _driver.getLastScheduledTaskExecutionInfo(queueName);
+ return (execInfo.getJobName().equals(queueName + "_job_0")
+ && execInfo.getTaskPartitionState() == TaskPartitionState.COMPLETED
+ && execInfo.getStartTimeStamp().equals(lastScheduledTaskTs)
+ && startTimesFastTasks.get(startTimesFastTasks.size() - 1).equals(lastScheduledTaskTs));
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(hasQueueReachedDesiredState);
}
/**
@@ -82,11 +105,12 @@ public class TestGetLastScheduledTaskExecInfo extends TaskTestBase {
* @param jobQueueName name of the queue
* @param numTasks number of tasks to schedule
* @param taskTimeout duration of each task to be run for
+ * @param expectedScheduledTasks expected number of tasks that should be scheduled
* @return list of timestamps for all tasks in ascending order
- * @throws InterruptedException
+ * @throws Exception
*/
- private List<Long> setupTasks(String jobQueueName, int numTasks, long taskTimeout)
- throws InterruptedException {
+ private List<Long> setupTasks(String jobQueueName, int numTasks, long taskTimeout,
+ int expectedScheduledTasks) throws Exception {
// Create a queue
JobQueue.Builder queueBuilder = TaskTestUtil.buildJobQueue(jobQueueName);
@@ -104,8 +128,26 @@ public class TestGetLastScheduledTaskExecInfo extends TaskTestBase {
jobConfig.addTaskConfigs(taskConfigs).setNumConcurrentTasksPerInstance(2);
queueBuilder.enqueueJob("job_0", jobConfig);
_driver.start(queueBuilder.build());
- // 1 second delay for the Controller
- Thread.sleep(1000);
+
+ _driver.pollForWorkflowState(jobQueueName, TaskState.IN_PROGRESS);
+
+ boolean haveExpectedNumberOfTasksScheduled = TestHelper.verify(() -> {
+ int scheduleTask = 0;
+ WorkflowConfig workflowConfig =
+ TaskUtil.getWorkflowConfig(_manager.getHelixDataAccessor(), jobQueueName);
+ for (String job : workflowConfig.getJobDag().getAllNodes()) {
+ JobContext jobContext = _driver.getJobContext(job);
+ Set<Integer> allPartitions = jobContext.getPartitionSet();
+ for (Integer partition : allPartitions) {
+ String timestamp = jobContext.getMapField(partition).get(TASK_START_TIME_KEY);
+ if (timestamp != null) {
+ scheduleTask++;
+ }
+ }
+ }
+ return (scheduleTask == expectedScheduledTasks);
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(haveExpectedNumberOfTasksScheduled);
// Pull jobContexts and look at the start times
List<Long> startTimes = new ArrayList<>();
diff --git a/helix-core/src/test/java/org/apache/helix/tools/TestClusterVerifier.java b/helix-core/src/test/java/org/apache/helix/tools/TestClusterVerifier.java
index 21e5935..47113b8 100644
--- a/helix-core/src/test/java/org/apache/helix/tools/TestClusterVerifier.java
+++ b/helix-core/src/test/java/org/apache/helix/tools/TestClusterVerifier.java
@@ -119,7 +119,7 @@ public class TestClusterVerifier extends ZkUnitTestBase {
}
@Test
- public void testDisablePartitionAndStopInstance() throws InterruptedException {
+ public void testDisablePartitionAndStopInstance() throws Exception {
// Just ensure that the entire cluster passes
// ensure that the external view coalesces
HelixClusterVerifier bestPossibleVerifier =
@@ -133,8 +133,14 @@ public class TestClusterVerifier extends ZkUnitTestBase {
// Disable partition for 1 instance, then Full-Auto ExternalView should not match IdealState.
_admin.enablePartition(false, _clusterName, _participants[0].getInstanceName(), FULL_AUTO_RESOURCES[0],
Lists.newArrayList(FULL_AUTO_RESOURCES[0] + "_0"));
- Thread.sleep(1000);
- Assert.assertFalse(strictMatchVerifier.verify(3000));
+
+ boolean isVerifiedFalse = TestHelper.verify(() -> {
+ HelixClusterVerifier strictMatchVerifierTemp =
+ new StrictMatchExternalViewVerifier.Builder(_clusterName).setZkClient(_gZkClient).build();
+ boolean verified = strictMatchVerifierTemp.verify(3000);
+ return (!verified);
+ }, TestHelper.WAIT_DURATION);
+ Assert.assertTrue(isVerifiedFalse);
// Enable the partition back
_admin.enablePartition(true, _clusterName, _participants[0].getInstanceName(), FULL_AUTO_RESOURCES[0],
@@ -174,11 +180,16 @@ public class TestClusterVerifier extends ZkUnitTestBase {
// Ensure that this passes even when one resource is down
_admin.enableInstance(_clusterName, "localhost_12918", false);
- Thread.sleep(1000);
+
+ ZkHelixClusterVerifier verifier =
+ new BestPossibleExternalViewVerifier.Builder(_clusterName).setZkClient(_gZkClient)
+ .setResources(Sets.newHashSet(testDB)).build();
+ Assert.assertTrue(verifier.verifyByPolling());
+
_admin.enableCluster(_clusterName, false);
_admin.enableInstance(_clusterName, "localhost_12918", true);
- ZkHelixClusterVerifier verifier =
+ verifier =
new BestPossibleExternalViewVerifier.Builder(_clusterName).setZkClient(_gZkClient)
.setResources(Sets.newHashSet(testDB)).build();
Assert.assertTrue(verifier.verifyByPolling());