helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From l...@apache.org
Subject [12/33] helix git commit: Add static methods into TaskDriver for getting configuration/context for jobs and workflows.
Date Wed, 17 Aug 2016 04:27:08 GMT
Add static methods into TaskDriver for getting configuration/context for jobs and workflows.


Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/aeb6f3ec
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/aeb6f3ec
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/aeb6f3ec

Branch: refs/heads/helix-0.6.x
Commit: aeb6f3ec7ab973316fd3468e9cbb0052a6a4306e
Parents: d386aff
Author: Lei Xia <lxia@linkedin.com>
Authored: Mon Mar 21 13:46:01 2016 -0700
Committer: Lei Xia <lxia@linkedin.com>
Committed: Tue Jul 5 14:53:34 2016 -0700

----------------------------------------------------------------------
 .../webapp/resources/JobQueueResource.java      | 19 +++++++------------
 .../helix/webapp/resources/JobResource.java     | 11 ++---------
 .../java/org/apache/helix/task/TaskDriver.java  | 16 ++++++++++++++++
 .../java/org/apache/helix/task/TaskUtil.java    | 20 ++++++++++----------
 .../integration/task/TestTaskRebalancer.java    |  2 --
 5 files changed, 35 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/helix/blob/aeb6f3ec/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobQueueResource.java
----------------------------------------------------------------------
diff --git a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobQueueResource.java
b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobQueueResource.java
index 830e16b..954ae73 100644
--- a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobQueueResource.java
+++ b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobQueueResource.java
@@ -21,15 +21,10 @@ package org.apache.helix.webapp.resources;
 
 import org.apache.helix.HelixDataAccessor;
 import org.apache.helix.HelixException;
-import org.apache.helix.HelixProperty;
 import org.apache.helix.PropertyKey;
-import org.apache.helix.PropertyPathConfig;
-import org.apache.helix.PropertyType;
 import org.apache.helix.ZNRecord;
-import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.manager.zk.ZkClient;
-import org.apache.helix.store.HelixPropertyStore;
-import org.apache.helix.store.zk.ZkHelixPropertyStore;
+import org.apache.helix.model.ResourceConfig;
 import org.apache.helix.task.JobConfig;
 import org.apache.helix.task.TaskDriver;
 import org.apache.helix.task.TaskUtil;
@@ -91,14 +86,14 @@ public class JobQueueResource extends ServerResource {
         ClusterRepresentationUtil.getClusterDataAccessor(zkClient, clusterName);
     PropertyKey.Builder keyBuilder = accessor.keyBuilder();
 
-    // Get job queue config
-    HelixProperty jobQueueConfig = accessor.getProperty(keyBuilder.resourceConfig(jobQueueName));
+    TaskDriver taskDriver = new TaskDriver(zkClient, clusterName);
 
+    // Get job queue config
+    // TODO: fix this to use workflowConfig.
+    ResourceConfig jobQueueConfig = accessor.getProperty(keyBuilder.resourceConfig(jobQueueName));
+    
     // Get job queue context
-    String path = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, clusterName);
-    HelixPropertyStore<ZNRecord> propertyStore =
-        new ZkHelixPropertyStore<ZNRecord>(new ZkBaseDataAccessor<ZNRecord>(zkClient),
path, null);
-    WorkflowContext ctx = TaskUtil.getWorkflowContext(propertyStore, jobQueueName);
+    WorkflowContext ctx = taskDriver.getWorkflowContext(jobQueueName);
 
     // Create the result
     ZNRecord hostedEntitiesRecord = new ZNRecord(jobQueueName);

http://git-wip-us.apache.org/repos/asf/helix/blob/aeb6f3ec/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobResource.java
----------------------------------------------------------------------
diff --git a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobResource.java
b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobResource.java
index cdcde35..d31c81b 100644
--- a/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobResource.java
+++ b/helix-admin-webapp/src/main/java/org/apache/helix/webapp/resources/JobResource.java
@@ -25,10 +25,7 @@ import org.apache.helix.PropertyKey;
 import org.apache.helix.PropertyPathConfig;
 import org.apache.helix.PropertyType;
 import org.apache.helix.ZNRecord;
-import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.manager.zk.ZkClient;
-import org.apache.helix.store.HelixPropertyStore;
-import org.apache.helix.store.zk.ZkHelixPropertyStore;
 import org.apache.helix.task.JobContext;
 import org.apache.helix.task.TaskDriver;
 import org.apache.helix.task.TaskUtil;
@@ -119,14 +116,10 @@ public class JobResource extends ServerResource {
     // Get job queue config
     String namespacedJobName = TaskUtil.getNamespacedJobName(jobQueueName, jobName);
     HelixProperty jobConfig = accessor.getProperty(keyBuilder.resourceConfig(namespacedJobName));
+    TaskDriver taskDriver = new TaskDriver(zkClient, clusterName);
 
     // Get job queue context
-    JobContext ctx = null;
-    String path = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, clusterName);
-    HelixPropertyStore<ZNRecord> propertyStore =
-        new ZkHelixPropertyStore<ZNRecord>(new ZkBaseDataAccessor<ZNRecord>(zkClient),
path, null);
-
-    ctx = TaskUtil.getJobContext(propertyStore, namespacedJobName);
+    JobContext ctx = taskDriver.getJobContext(namespacedJobName);
 
     // Create the result
     ZNRecord hostedEntitiesRecord = new ZNRecord(namespacedJobName);

http://git-wip-us.apache.org/repos/asf/helix/blob/aeb6f3ec/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java b/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
index 99bcb62..c0d7852 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
@@ -733,6 +733,22 @@ public class TaskDriver {
     return TaskUtil.getJobContext(_propertyStore, job);
   }
 
+  public static JobContext getJobContext(HelixManager manager, String job) {
+    return TaskUtil.getJobContext(manager, job);
+  }
+
+  public static WorkflowConfig getWorkflowConfig(HelixManager manager, String workflow) {
+    return TaskUtil.getWorkflowCfg(manager, workflow);
+  }
+
+  public static WorkflowContext getWorkflowContext(HelixManager manager, String workflow)
{
+    return TaskUtil.getWorkflowContext(manager, workflow);
+  }
+
+  public static JobConfig getJobConfig(HelixManager manager, String job) {
+    return TaskUtil.getJobCfg(manager, job);
+  }
+
   public void list(String resource) {
     WorkflowConfig wCfg = TaskUtil.getWorkflowCfg(_accessor, resource);
     if (wCfg == null) {

http://git-wip-us.apache.org/repos/asf/helix/blob/aeb6f3ec/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java b/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
index 513c14e..8745a82 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
@@ -51,7 +51,7 @@ public class TaskUtil {
 
   /**
    * Parses job resource configurations in Helix into a {@link JobConfig} object.
-   * This method is internal API, please use TaskDriver.getJobConfig();
+   * This method is internal API, please use the corresponding one in TaskDriver.getJobConfig();
    *
    * @param accessor    Accessor to access Helix configs
    * @param jobResource The name of the job resource
@@ -78,7 +78,7 @@ public class TaskUtil {
 
   /**
    * Parses job resource configurations in Helix into a {@link JobConfig} object.
-   * This method is internal API, please use TaskDriver.getJobConfig();
+   * This method is internal API, please use the corresponding one in TaskDriver.getJobConfig();
    *
    * @param manager     HelixManager object used to connect to Helix.
    * @param jobResource The name of the job resource.
@@ -91,7 +91,7 @@ public class TaskUtil {
 
   /**
    * Parses workflow resource configurations in Helix into a {@link WorkflowConfig} object.
-   * This method is internal API, please use TaskDriver.getWorkflowConfig();
+   * This method is internal API, please use the corresponding one in TaskDriver.getWorkflowConfig();
    *
    * @param accessor  Accessor to access Helix configs
    * @param workflow The name of the workflow.
@@ -112,7 +112,7 @@ public class TaskUtil {
 
   /**
    * Parses workflow resource configurations in Helix into a {@link WorkflowConfig} object.
-   * This method is internal API, please use TaskDriver.getWorkflowConfig();
+   * This method is internal API, please use the corresponding one in TaskDriver.getWorkflowConfig();
    *
    * @param manager          Helix manager object used to connect to Helix.
    * @param workflow The name of the workflow resource.
@@ -143,7 +143,7 @@ public class TaskUtil {
    * @param jobResource   The name of the job
    * @return the {@link JobContext}, or null if none is available
    */
-  public static JobContext getJobContext(HelixPropertyStore<ZNRecord> propertyStore,
+  protected static JobContext getJobContext(HelixPropertyStore<ZNRecord> propertyStore,
       String jobResource) {
     ZNRecord r = propertyStore
         .get(Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, jobResource, CONTEXT_NODE),
@@ -179,13 +179,13 @@ public class TaskUtil {
 
   /**
    * Get the runtime context of a single workflow.
-   * This method is internal API, please use TaskDriver.getWorkflowContext();
+   * This method is internal API, please use the corresponding one in TaskDriver.getWorkflowContext();
    *
    * @param propertyStore    Property store of the cluster
    * @param workflowResource The name of the workflow
    * @return the {@link WorkflowContext}, or null if none is available
    */
-  public static WorkflowContext getWorkflowContext(HelixPropertyStore<ZNRecord> propertyStore,
+  protected static WorkflowContext getWorkflowContext(HelixPropertyStore<ZNRecord>
propertyStore,
       String workflowResource) {
     ZNRecord r = propertyStore.get(
         Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowResource, CONTEXT_NODE),
@@ -195,7 +195,7 @@ public class TaskUtil {
 
   /**
    * Get the runtime context of a single workflow.
-   * This method is internal API, please use TaskDriver.getWorkflowContext();
+   * This method is internal API, please use the corresponding one in TaskDriver.getWorkflowContext();
    *
    * @param manager          a connection to Helix
    * @param workflowResource the name of the workflow
@@ -212,7 +212,7 @@ public class TaskUtil {
    * @param workflowResource the name of the workflow
    * @param ctx              the up-to-date {@link WorkflowContext} for the workflow
    */
-  public static void setWorkflowContext(HelixManager manager, String workflowResource,
+  protected static void setWorkflowContext(HelixManager manager, String workflowResource,
       WorkflowContext ctx) {
     manager.getHelixPropertyStore().set(
         Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowResource, CONTEXT_NODE),
@@ -298,7 +298,7 @@ public class TaskUtil {
    * @param accessor Helix data accessor
    * @param resource the name of the resource changed to triggering the execution
    */
-  public static void invokeRebalance(HelixDataAccessor accessor, String resource) {
+  protected static void invokeRebalance(HelixDataAccessor accessor, String resource) {
     // The pipeline is idempotent, so touching an ideal state is enough to trigger a pipeline
run
     LOG.info("invoke rebalance for " + resource);
     PropertyKey key = accessor.keyBuilder().idealStates(resource);

http://git-wip-us.apache.org/repos/asf/helix/blob/aeb6f3ec/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
index 3a5b179..74b10fd 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancer.java
@@ -184,8 +184,6 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
     // Wait for job to finish and expire
     TaskTestUtil.pollForWorkflowState(_driver, jobName, TaskState.COMPLETED);
     Thread.sleep(expiry);
-    TaskUtil.invokeRebalance(_manager.getHelixDataAccessor(), flow.getName());
-    Thread.sleep(expiry);
 
     // Ensure workflow config and context were cleaned up by now
     Assert.assertFalse(_manager.getHelixPropertyStore().exists(workflowPropStoreKey,


Mime
View raw message