You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by GitBox <gi...@apache.org> on 2020/10/05 20:34:18 UTC
[GitHub] [helix] alirezazamani commented on a change in pull request #1227: Implement thread leakage check, fixing major thread leakage test and some flaky test
alirezazamani commented on a change in pull request #1227:
URL: https://github.com/apache/helix/pull/1227#discussion_r499848087
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestDeleteJobFromJobQueue.java
##########
@@ -69,11 +70,16 @@ public void testForceDeleteJobFromJobQueue() throws InterruptedException {
Assert
.assertNotNull(_driver.getJobContext(TaskUtil.getNamespacedJobName(jobQueueName, "job2")));
- // The following force delete for the job should go through without getting an exception
- _driver.deleteJob(jobQueueName, "job2", true);
+ try {
Review comment:
Another way can be to stop the controller and do force delete? Is there any reason you chose to use try-catch? I think stopping the controller and doing force delete can be a better approach as it actually checks the dele operation.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -231,12 +272,14 @@ public void testQueueJobsMaxCapacity() throws InterruptedException {
_driver.resume(queueName);
+ System.out.println("before the polls");
Review comment:
Please check the whole PR and remove these prints is possible.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
##########
@@ -102,29 +123,41 @@ public void stopAndResumeNamedQueue() throws Exception {
String queueName = TestHelper.getTestMethodName();
// Create a queue
- LOG.info("Starting job-queue: " + queueName);
- JobQueue queue = new JobQueue.Builder(queueName).build();
+ LOG.info("Starting job-queue : " + queueName);
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
+ WorkflowConfig wfg = wfgBuilder.build();
+ JobQueue queue = new JobQueue.Builder(queueName)
+ .setWorkflowConfig(wfg)
+ .build();
_driver.createQueue(queue);
// Enqueue jobs
+ List<String> jobNames = new ArrayList<>();
+ List<JobConfig.Builder> jobBuilders = new ArrayList<>();
Set<String> master = Sets.newHashSet("MASTER");
JobConfig.Builder job1 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(master)
.setJobCommandConfigMap(Collections.singletonMap(MockTask.JOB_DELAY, "200"));
String job1Name = "masterJob";
LOG.info("Enqueuing job: " + job1Name);
- _driver.enqueueJob(queueName, job1Name, job1);
+ jobNames.add(job1Name);
+ jobBuilders.add(job1);
+ //_driver.enqueueJob(queueName, job1Name, job1);
Review comment:
remove?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -102,7 +136,9 @@ public void testJobSubmitGenericWorkflows() throws InterruptedException {
JobConfig.Builder jobBuilder =
new JobConfig.Builder().setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB)
.setCommand(MockTask.TASK_COMMAND).setMaxAttemptsPerTask(2);
- Workflow.Builder builder = new Workflow.Builder(workflowName);
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
Review comment:
nit wfBuilder.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -65,14 +97,16 @@ public void testJobQueueAddingJobsOneByOne() throws InterruptedException {
TaskState.COMPLETED);
}
- @Test
+ @Test()
public void testJobQueueAddingJobsAtSametime() throws InterruptedException {
String queueName = TestHelper.getTestMethodName();
JobQueue.Builder builder = TaskTestUtil.buildJobQueue(queueName);
WorkflowConfig.Builder workflowCfgBuilder =
- new WorkflowConfig.Builder().setWorkflowId(queueName).setParallelJobs(1);
+ new WorkflowConfig.Builder().setWorkflowId(queueName).setParallelJobs(1)
+ .setJobPurgeInterval(-1);
_driver.start(builder.setWorkflowConfig(workflowCfgBuilder.build()).build());
+ System.out.println("before add jobs");
Review comment:
You can delete this line?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -188,6 +227,7 @@ public void testQueueParallelJobs() throws InterruptedException {
TaskState.COMPLETED);
}
+ System.out.println("after the poll");
Review comment:
same.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -165,6 +202,7 @@ public void testQueueParallelJobs() throws InterruptedException {
TaskState.COMPLETED);
}
+ System.out.println("Before stop controller");
Review comment:
I don't think these prints are necessary.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -41,13 +47,39 @@
public void beforeClass() throws Exception {
setSingleTestEnvironment();
super.beforeClass();
+ //WorkflowConfig.disableJobPurge();
+ }
+
+ @AfterClass
+ @Override
+ public void afterClass() throws Exception {
+ //WorkflowConfig.enableJobPurge();
+ super.afterClass();
+ }
+
+ @BeforeMethod
+ public void beforeTest(Method testMethod, ITestContext testContext) {
+ long startTime = System.currentTimeMillis();
+ System.out.println("START " + testMethod.getName() + " at " + new Date(startTime));
+ testContext.setAttribute("StartTime", System.currentTimeMillis());
+ }
+
+ @AfterMethod
+ public void endTest(Method testMethod, ITestContext testContext) {
+ Long startTime = (Long) testContext.getAttribute("StartTime");
+ long endTime = System.currentTimeMillis();
+ System.out.println("END " + testMethod.getName() + " at " + new Date(endTime) + ", took: "
+ + (endTime - startTime) + "ms.");
}
@Test
public void testJobQueueAddingJobsOneByOne() throws InterruptedException {
String queueName = TestHelper.getTestMethodName();
JobQueue.Builder builder = TaskTestUtil.buildJobQueue(queueName);
- WorkflowConfig.Builder workflowCfgBuilder = new WorkflowConfig.Builder().setWorkflowId(queueName).setParallelJobs(1);
+ WorkflowConfig.Builder workflowCfgBuilder = new WorkflowConfig.Builder()
+ .setWorkflowId(queueName)
+ .setParallelJobs(1)
+ .setJobPurgeInterval(-1);
Review comment:
As we discussed offline, if you are sure that this method does not cause overwriting the old configs, then this is fine with me.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -177,6 +215,7 @@ public void testQueueParallelJobs() throws InterruptedException {
}
_driver.enqueueJobs(queueName, jobNames, jobBuilders);
+ System.out.println("before start controller");
Review comment:
same here.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -92,6 +125,7 @@ public void testJobQueueAddingJobsAtSametime() throws InterruptedException {
_driver.resume(queueName);
+ System.out.println("before poll to job state");
Review comment:
delete?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -247,6 +290,7 @@ public void testQueueJobsMaxCapacity() throws InterruptedException {
}
Assert.assertTrue(exceptionHappenedWhileAddingNewJob);
+ System.out.println("after the exception");
Review comment:
same.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestEnqueueJobs.java
##########
@@ -231,12 +272,14 @@ public void testQueueJobsMaxCapacity() throws InterruptedException {
_driver.resume(queueName);
+ System.out.println("before the polls");
// Wait until all of the enqueued jobs (Job0 to Job3) are finished
for (int i = 0; i < numberOfJobsAddedInitially; i++) {
_driver.pollForJobState(queueName, TaskUtil.getNamespacedJobName(queueName, "JOB" + i),
TaskState.COMPLETED);
}
+ System.out.println("before enqueue for exception");
Review comment:
print?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestStuckTaskQuota.java
##########
@@ -63,6 +63,9 @@ public void beforeClass() throws Exception {
@AfterClass
public void afterClass() throws Exception {
+ String testClassName = this.getShortClassName();
+ System.out.println("AfterClass: " + testClassName + " of TestStuckTaskQuota called.");
Review comment:
Same here. You can use methods to get the names here.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestQuotaBasedScheduling.java
##########
@@ -543,11 +561,17 @@ public void testJobQueueScheduling() throws InterruptedException {
jobConfigBulider = new JobConfig.Builder().setCommand(JOB_COMMAND).addTaskConfigs(taskConfigs)
.setJobCommandConfigMap(_jobCommandMap).setJobType("B");
+ jobNames.clear();
+ jobBuilders.clear();
for (int i = 5; i < 10; i++) {
String jobName = "JOB_" + i;
lastJobName = jobName;
- _driver.enqueueJob(queueName, jobName, jobConfigBulider);
+ jobNames.add(jobName);
+ jobBuilders.add(jobConfigBulider);
+ //_driver.enqueueJob(queueName, jobName, jobConfigBulider);
Review comment:
remove this line?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestMaxNumberOfAttemptsMasterSwitch.java
##########
@@ -71,6 +71,9 @@ public void beforeClass() throws Exception {
@AfterClass
public void afterClass() throws Exception {
+ String testClassName = this.getShortClassName();
+ System.out.println("AfterClass: " + testClassName + " of TestMaxNumberOfAttempsMasterSwitch called.");
Review comment:
Can you use TestHelper.getTestMethodName() and TestHelper.getTestClassName?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
##########
@@ -217,22 +246,35 @@ public void testNamedQueue() throws Exception {
String queueName = TestHelper.getTestMethodName();
// Create a queue
- JobQueue queue = new JobQueue.Builder(queueName).build();
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
+ WorkflowConfig wfg = wfgBuilder.build();
+ JobQueue queue = new JobQueue.Builder(queueName)
+ .setWorkflowConfig(wfg)
+ .build();
_driver.createQueue(queue);
// Enqueue jobs
+ List<String> jobNames = new ArrayList<>();
+ List<JobConfig.Builder> jobBuilders = new ArrayList<>();
Set<String> master = Sets.newHashSet("MASTER");
Set<String> slave = Sets.newHashSet("SLAVE");
JobConfig.Builder job1 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(master);
JobConfig.Builder job2 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(slave);
- _driver.enqueueJob(queueName, "masterJob", job1);
- _driver.enqueueJob(queueName, "slaveJob", job2);
+ jobNames.add("masterJob");
+ jobNames.add("slaveJob");
+ jobBuilders.add(job1);
+ jobBuilders.add(job2);
+ //_driver.enqueueJob(queueName, "masterJob", job1);
Review comment:
remove the comments if possible.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestQuotaBasedScheduling.java
##########
@@ -526,11 +538,17 @@ public void testJobQueueScheduling() throws InterruptedException {
JobConfig.Builder jobConfigBulider = new JobConfig.Builder().setCommand(JOB_COMMAND)
.addTaskConfigs(taskConfigs).setJobCommandConfigMap(_jobCommandMap).setJobType("A");
+ List<String> jobNames = new ArrayList<>();
+ List<JobConfig.Builder> jobBuilders = new ArrayList<>();
+
for (int i = 0; i < 5; i++) {
String jobName = "JOB_" + i;
lastJobName = jobName;
- _driver.enqueueJob(queueName, jobName, jobConfigBulider);
+ jobNames.add(jobName);
+ jobBuilders.add(jobConfigBulider);
+ // _driver.enqueueJob(queueName, jobName, jobConfigBulider);
Review comment:
Remove this line?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
##########
@@ -162,8 +183,12 @@ public void partitionSet() throws Exception {
@Test
public void testRepeatedWorkflow() throws Exception {
String workflowName = "SomeWorkflow";
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
+ WorkflowConfig wfg = wfgBuilder.build();
Review comment:
nit, wgBuilder
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestJobFailureDependence.java
##########
@@ -161,15 +161,19 @@ public void testWorkflowFailureJobThreshold() throws Exception {
_driver.updateWorkflow(queueName, configBuilder.build());
_driver.stop(queueName);
+ List<String> jobNames = new ArrayList<>();
+ List<JobConfig.Builder> jobBuilders = new ArrayList<>();
for (int i = 0; i < _numDbs; i++) {
JobConfig.Builder jobConfig =
new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND).setTargetResource(_testDbs.get(i))
.setTargetPartitionStates(Sets.newHashSet("SLAVE")).setIgnoreDependentJobFailure(true);
String jobName = "job" + _testDbs.get(i);
queueBuilder.enqueueJob(jobName, jobConfig);
- _driver.enqueueJob(queueName, jobName, jobConfig);
+ jobNames.add(jobName);
+ jobBuilders.add(jobConfig);
+ //_driver.enqueueJob(queueName, jobName, jobConfig);
Review comment:
You can delete this commented line.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
##########
@@ -217,22 +246,35 @@ public void testNamedQueue() throws Exception {
String queueName = TestHelper.getTestMethodName();
// Create a queue
- JobQueue queue = new JobQueue.Builder(queueName).build();
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
+ WorkflowConfig wfg = wfgBuilder.build();
+ JobQueue queue = new JobQueue.Builder(queueName)
+ .setWorkflowConfig(wfg)
+ .build();
_driver.createQueue(queue);
// Enqueue jobs
+ List<String> jobNames = new ArrayList<>();
+ List<JobConfig.Builder> jobBuilders = new ArrayList<>();
Set<String> master = Sets.newHashSet("MASTER");
Set<String> slave = Sets.newHashSet("SLAVE");
JobConfig.Builder job1 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(master);
JobConfig.Builder job2 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(slave);
- _driver.enqueueJob(queueName, "masterJob", job1);
- _driver.enqueueJob(queueName, "slaveJob", job2);
+ jobNames.add("masterJob");
+ jobNames.add("slaveJob");
+ jobBuilders.add(job1);
+ jobBuilders.add(job2);
+ //_driver.enqueueJob(queueName, "masterJob", job1);
+ //_driver.enqueueJob(queueName, "slaveJob", job2);
+ _driver.enqueueJobs(queueName, jobNames, jobBuilders);
// Ensure successful completion
String namespacedJob1 = queueName + "_masterJob";
String namespacedJob2 = queueName + "_slaveJob";
+
+ System.out.println("testNamedQueue before pollForJobState");
Review comment:
remove prints if possible.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
##########
@@ -64,7 +78,10 @@ public void testExpiry() throws Exception {
JobConfig.Builder jobBuilder = JobConfig.Builder.fromMap(WorkflowGenerator.DEFAULT_JOB_CONFIG);
jobBuilder.setJobCommandConfigMap(commandConfig);
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
Review comment:
wfBuilder?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerParallel.java
##########
@@ -109,16 +117,22 @@ public void testWhenAllowOverlapJobAssignment() throws Exception {
for (int i = 0; i < PARALLEL_COUNT; i++) {
List<TaskConfig> taskConfigs = new ArrayList<TaskConfig>();
for (int j = 0; j < TASK_COUNT; j++) {
- taskConfigs.add(new TaskConfig.Builder().setTaskId("job_" + (i + 1) + "_task_" + j)
- .setCommand(MockTask.TASK_COMMAND).build());
+ taskConfigs.add(new TaskConfig.Builder()
+ .setTaskId("job_" + (i + 1) + "_task_" + j)
+ .setCommand(MockTask.TASK_COMMAND)
+ .addConfig(MockTask.JOB_DELAY, "2000")
+ .build());
}
jobConfigBuilders.add(new JobConfig.Builder().addTaskConfigs(taskConfigs));
}
_driver.stop(queueName);
+ List<String> jobNames = new ArrayList<>();
for (int i = 0; i < jobConfigBuilders.size(); ++i) {
- _driver.enqueueJob(queueName, "job_" + (i + 1), jobConfigBuilders.get(i));
+ jobNames.add("job_" + (i + 1));
+ //_driver.enqueueJob(queueName, "job_" + (i + 1), jobConfigBuilders.get(i));
Review comment:
remove?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskSchedulingTwoCurrentStates.java
##########
@@ -93,6 +93,8 @@ public void beforeClass() throws Exception {
@AfterClass()
public void afterClass() throws Exception {
+ String testClassName = this.getShortClassName();
+ System.out.println("AfterClass: " + testClassName + " of TestTaskSchedulingTwoCurrentStates called.");
Review comment:
Same here. You can use TestHelper methods.
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskThrottling.java
##########
@@ -87,15 +97,17 @@ public void testTaskThrottle() throws Exception {
_gSetupTool.getClusterManagementTool().setConfig(scope, properties);
String jobName2 = "Job2";
- Workflow flow2 = WorkflowGenerator.generateSingleJobWorkflowBuilder(jobName2, jobConfig).build();
+ Workflow flow2 = WorkflowGenerator.generateSingleJobWorkflowBuilder(jobName2, jobConfig)
+ .setWorkflowConfig(wfg)
+ .build();
_driver.start(flow2);
_driver.pollForJobState(flow2.getName(), TaskUtil.getNamespacedJobName(flow2.getName(), jobName2),
TaskState.IN_PROGRESS);
// Expect 10 tasks
Assert.assertTrue(TestHelper.verify(
() -> (countRunningPartition(flow2, jobName2) == (_numNodes * perNodeTaskLimitation)),
- TestHelper.WAIT_DURATION));
+ 2 * TestHelper.WAIT_DURATION));
Review comment:
Now even when you change it to 60 seconds, is this 2 * necessary?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
##########
@@ -102,29 +123,41 @@ public void stopAndResumeNamedQueue() throws Exception {
String queueName = TestHelper.getTestMethodName();
// Create a queue
- LOG.info("Starting job-queue: " + queueName);
- JobQueue queue = new JobQueue.Builder(queueName).build();
+ LOG.info("Starting job-queue : " + queueName);
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
+ WorkflowConfig wfg = wfgBuilder.build();
+ JobQueue queue = new JobQueue.Builder(queueName)
+ .setWorkflowConfig(wfg)
+ .build();
_driver.createQueue(queue);
// Enqueue jobs
+ List<String> jobNames = new ArrayList<>();
+ List<JobConfig.Builder> jobBuilders = new ArrayList<>();
Set<String> master = Sets.newHashSet("MASTER");
JobConfig.Builder job1 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(master)
.setJobCommandConfigMap(Collections.singletonMap(MockTask.JOB_DELAY, "200"));
String job1Name = "masterJob";
LOG.info("Enqueuing job: " + job1Name);
- _driver.enqueueJob(queueName, job1Name, job1);
+ jobNames.add(job1Name);
+ jobBuilders.add(job1);
+ //_driver.enqueueJob(queueName, job1Name, job1);
Set<String> slave = Sets.newHashSet("SLAVE");
JobConfig.Builder job2 = new JobConfig.Builder().setCommand(MockTask.TASK_COMMAND)
.setTargetResource(WorkflowGenerator.DEFAULT_TGT_DB).setTargetPartitionStates(slave);
String job2Name = "slaveJob";
LOG.info("Enqueuing job: " + job2Name);
- _driver.enqueueJob(queueName, job2Name, job2);
+ jobNames.add(job2Name);
+ jobBuilders.add(job2);
+ _driver.enqueueJobs(queueName, jobNames,jobBuilders);
+ //_driver.enqueueJob(queueName, job2Name, job2);
Review comment:
remove?
##########
File path: helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
##########
@@ -82,7 +99,11 @@ public void stopAndResume() throws Exception {
@Test
public void stopAndResumeWorkflow() throws Exception {
String workflow = "SomeWorkflow";
- Workflow flow = WorkflowGenerator.generateDefaultRepeatedJobWorkflowBuilder(workflow).build();
+ WorkflowConfig.Builder wfgBuilder = new WorkflowConfig.Builder().setJobPurgeInterval(-1);
+ WorkflowConfig wfg = wfgBuilder.build();
Review comment:
Same here.
##########
File path: helix-core/src/test/java/org/apache/helix/task/TaskSynchronizedTestBase.java
##########
@@ -76,11 +77,15 @@ public void beforeClass() throws Exception {
startParticipants();
createManagers();
_clusterVerifier =
- new BestPossibleExternalViewVerifier.Builder(CLUSTER_NAME).setZkClient(_gZkClient).build();
+ new BestPossibleExternalViewVerifier.Builder(CLUSTER_NAME).setZkClient(_gZkClient)
+ .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME)
+ .build();
}
@AfterClass
public void afterClass() throws Exception {
+ String testClassName = this.getShortClassName();
+ System.out.println("AfterClass: " + testClassName + " of TaskSynchronizedTestBase called.");
Review comment:
TestHelper methods/
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org