stable test

Author: xyuanlu

helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java

@@ -30,14 +30,14 @@
 import java.util.TreeSet;
 
 import com.google.common.collect.ImmutableMap;
-import org.apache.helix.util.RebalanceUtil;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.apache.helix.controller.dataproviders.WorkflowControllerDataProvider;
 import org.apache.helix.controller.stages.CurrentStateOutput;
 import org.apache.helix.model.Message;
 import org.apache.helix.model.Partition;
 import org.apache.helix.model.ResourceAssignment;
 import org.apache.helix.task.assigner.ThreadCountBasedTaskAssigner;
+import org.apache.helix.util.RebalanceUtil;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,21 +57,21 @@ public ResourceAssignment processJobStatusUpdateAndAssignment(String jobName,
       CurrentStateOutput currStateOutput, WorkflowContext workflowCtx) {
     // Fetch job configuration
     final JobConfig jobCfg = _dataProvider.getJobConfig(jobName);
+    String workflowResource = jobCfg.getWorkflow();
     if (jobCfg == null) {
-      LOG.error("Job configuration is NULL for " + jobName);
+      LOG.error("Job configuration is NULL for workflow " + workflowResource + " job " + jobName);
       return buildEmptyAssignment(jobName, currStateOutput);
     }
-    String workflowResource = jobCfg.getWorkflow();
 
     // Fetch workflow configuration and context
     final WorkflowConfig workflowCfg = _dataProvider.getWorkflowConfig(workflowResource);
     if (workflowCfg == null) {
-      LOG.error("Workflow configuration is NULL for " + jobName);
+      LOG.error("Workflow configuration is NULL for workflow " + workflowResource + " job " + jobName);
       return buildEmptyAssignment(jobName, currStateOutput);
     }
 
     if (workflowCtx == null) {
-      LOG.error("Workflow context is NULL for " + jobName);
+      LOG.error("Workflow context is NULL for workflow " + workflowResource + " job " + jobName);
       return buildEmptyAssignment(jobName, currStateOutput);
     }
 
@@ -103,7 +103,8 @@ public ResourceAssignment processJobStatusUpdateAndAssignment(String jobName,
     }
 
     if (!isWorkflowReadyForSchedule(workflowCfg)) {
-      LOG.info("Job is not ready to be run since workflow is not ready " + jobName);
+      LOG.info("Job is not ready to be run since workflow " + workflowResource + " is not ready "
+          + jobName);
       return buildEmptyAssignment(jobName, currStateOutput);
     }
 
@@ -145,7 +146,7 @@ public ResourceAssignment processJobStatusUpdateAndAssignment(String jobName,
             : _dataProvider.getEnabledLiveInstancesWithTag(jobCfg.getInstanceGroupTag());
 
     if (liveInstances.isEmpty()) {
-      LOG.error("No available instance found for job!");
+      LOG.error("No available instance found for job " + jobName);
     }
 
     TargetState jobTgtState = workflowCfg.getTargetState();

helix-core/src/test/java/org/apache/helix/integration/TestZkCallbackHandlerLeak.java

@@ -472,6 +472,7 @@ public void testCurrentStatePathLeakingByAsycRemoval() throws Exception {
       jobAccesor.setProperty(jobKey, cs);
     }
 
+    Thread.sleep(5000);
     Map<String, Set<String>> listenersByZkPath = ZkTestHelper.getListenersByZkPath(ZK_ADDR);
     Assert.assertTrue(listenersByZkPath.keySet().contains(jobKey.getPath()));
     rpWatchPaths = ZkTestHelper.getZkWatch(rpManager.getZkClient());