You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ka...@apache.org on 2014/04/30 20:39:33 UTC
[1/3] Port recent task framework changes
Repository: helix
Updated Branches:
refs/heads/helix-provisioning 8f0b7e4c6 -> 97ca4de4a
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/Workflow.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/Workflow.java b/helix-core/src/main/java/org/apache/helix/task/Workflow.java
index 902f616..5b27fb6 100644
--- a/helix-core/src/main/java/org/apache/helix/task/Workflow.java
+++ b/helix-core/src/main/java/org/apache/helix/task/Workflow.java
@@ -19,45 +19,53 @@ package org.apache.helix.task;
* under the License.
*/
-import com.google.common.base.Joiner;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.Reader;
import java.io.StringReader;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+
+import org.apache.helix.task.beans.JobBean;
import org.apache.helix.task.beans.TaskBean;
import org.apache.helix.task.beans.WorkflowBean;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+
/**
- * Houses a task dag and config set to fully describe a task workflow
+ * Houses a job dag and config set to fully describe a job workflow
*/
public class Workflow {
/** Default workflow name, useful constant for single-node workflows */
- public static enum WorkflowEnum {
- UNSPECIFIED;
- }
+ public static final String UNSPECIFIED = "UNSPECIFIED";
/** Workflow name */
- private final String _name;
+ private String _name;
/** Holds workflow-level configurations */
- private final WorkflowConfig _workflowConfig;
+ private WorkflowConfig _workflowConfig;
+
+ /** Contains the per-job configurations for all jobs specified in the provided dag */
+ private Map<String, Map<String, String>> _jobConfigs;
- /** Contains the per-task configurations for all tasks specified in the provided dag */
- private final Map<String, Map<String, String>> _taskConfigs;
+ /** Containers the per-job configurations of all individually-specified tasks */
+ private Map<String, List<TaskConfig>> _taskConfigs;
/** Constructs and validates a workflow against a provided dag and config set */
private Workflow(String name, WorkflowConfig workflowConfig,
- Map<String, Map<String, String>> taskConfigs) {
+ Map<String, Map<String, String>> jobConfigs, Map<String, List<TaskConfig>> taskConfigs) {
_name = name;
_workflowConfig = workflowConfig;
+ _jobConfigs = jobConfigs;
_taskConfigs = taskConfigs;
-
validate();
}
@@ -65,13 +73,17 @@ public class Workflow {
return _name;
}
- public Map<String, Map<String, String>> getTaskConfigs() {
+ public Map<String, Map<String, String>> getJobConfigs() {
+ return _jobConfigs;
+ }
+
+ public Map<String, List<TaskConfig>> getTaskConfigs() {
return _taskConfigs;
}
public Map<String, String> getResourceConfigMap() throws Exception {
Map<String, String> cfgMap = new HashMap<String, String>();
- cfgMap.put(WorkflowConfig.DAG, _workflowConfig.getTaskDag().toJson());
+ cfgMap.put(WorkflowConfig.DAG, _workflowConfig.getJobDag().toJson());
cfgMap.put(WorkflowConfig.EXPIRY, String.valueOf(_workflowConfig.getExpiry()));
cfgMap.put(WorkflowConfig.TARGET_STATE, _workflowConfig.getTargetState().name());
@@ -97,19 +109,19 @@ public class Workflow {
*
* <pre>
* name: MyFlow
- * tasks:
- * - name : TaskA
+ * jobs:
+ * - name : JobA
* command : SomeTask
* ...
- * - name : TaskB
- * parents : [TaskA]
+ * - name : JobB
+ * parents : [JobA]
* command : SomeOtherTask
* ...
- * - name : TaskC
+ * - name : JobC
* command : AnotherTask
* ...
- * - name : TaskD
- * parents : [TaskB, TaskC]
+ * - name : JobD
+ * parents : [JobB, JobC]
* command : AnotherTask
* ...
* </pre>
@@ -126,37 +138,44 @@ public class Workflow {
WorkflowBean wf = (WorkflowBean) yaml.load(reader);
Builder builder = new Builder(wf.name);
- for (TaskBean task : wf.tasks) {
- if (task.name == null) {
- throw new IllegalArgumentException("A task must have a name.");
+ for (JobBean job : wf.jobs) {
+ if (job.name == null) {
+ throw new IllegalArgumentException("A job must have a name.");
}
- if (task.parents != null) {
- for (String parent : task.parents) {
- builder.addParentChildDependency(parent, task.name);
+ if (job.parents != null) {
+ for (String parent : job.parents) {
+ builder.addParentChildDependency(parent, job.name);
}
}
- builder.addConfig(task.name, TaskConfig.WORKFLOW_ID, wf.name);
- builder.addConfig(task.name, TaskConfig.COMMAND, task.command);
- if (task.commandConfig != null) {
- builder.addConfig(task.name, TaskConfig.COMMAND_CONFIG, task.commandConfig.toString());
+ builder.addConfig(job.name, JobConfig.WORKFLOW_ID, wf.name);
+ builder.addConfig(job.name, JobConfig.COMMAND, job.command);
+ if (job.jobConfigMap != null) {
+ builder.addConfig(job.name, JobConfig.JOB_CONFIG_MAP, job.jobConfigMap.toString());
+ }
+ builder.addConfig(job.name, JobConfig.TARGET_RESOURCE, job.targetResource);
+ if (job.targetPartitionStates != null) {
+ builder.addConfig(job.name, JobConfig.TARGET_PARTITION_STATES,
+ Joiner.on(",").join(job.targetPartitionStates));
}
- builder.addConfig(task.name, TaskConfig.TARGET_RESOURCE, task.targetResource);
- if (task.targetPartitionStates != null) {
- builder.addConfig(task.name, TaskConfig.TARGET_PARTITION_STATES,
- Joiner.on(",").join(task.targetPartitionStates));
+ if (job.targetPartitions != null) {
+ builder.addConfig(job.name, JobConfig.TARGET_PARTITIONS,
+ Joiner.on(",").join(job.targetPartitions));
}
- if (task.targetPartitions != null) {
- builder.addConfig(task.name, TaskConfig.TARGET_PARTITIONS,
- Joiner.on(",").join(task.targetPartitions));
+ builder.addConfig(job.name, JobConfig.MAX_ATTEMPTS_PER_TASK,
+ String.valueOf(job.maxAttemptsPerPartition));
+ builder.addConfig(job.name, JobConfig.NUM_CONCURRENT_TASKS_PER_INSTANCE,
+ String.valueOf(job.numConcurrentTasksPerInstance));
+ builder.addConfig(job.name, JobConfig.TIMEOUT_PER_TASK,
+ String.valueOf(job.timeoutPerPartition));
+ if (job.tasks != null) {
+ List<TaskConfig> taskConfigs = Lists.newArrayList();
+ for (TaskBean task : job.tasks) {
+ taskConfigs.add(TaskConfig.from(task));
+ }
+ builder.addTaskConfigs(job.name, taskConfigs);
}
- builder.addConfig(task.name, TaskConfig.MAX_ATTEMPTS_PER_PARTITION,
- String.valueOf(task.maxAttemptsPerPartition));
- builder.addConfig(task.name, TaskConfig.NUM_CONCURRENT_TASKS_PER_INSTANCE,
- String.valueOf(task.numConcurrentTasksPerInstance));
- builder.addConfig(task.name, TaskConfig.TIMEOUT_PER_PARTITION,
- String.valueOf(task.timeoutPerPartition));
}
return builder.build();
@@ -168,47 +187,78 @@ public class Workflow {
*/
public void validate() {
// validate dag and configs
- if (!_taskConfigs.keySet().containsAll(_workflowConfig.getTaskDag().getAllNodes())) {
+ if (!_jobConfigs.keySet().containsAll(_workflowConfig.getJobDag().getAllNodes())) {
throw new IllegalArgumentException("Nodes specified in DAG missing from config");
- } else if (!_workflowConfig.getTaskDag().getAllNodes().containsAll(_taskConfigs.keySet())) {
+ } else if (!_workflowConfig.getJobDag().getAllNodes().containsAll(_jobConfigs.keySet())) {
throw new IllegalArgumentException("Given DAG lacks nodes with supplied configs");
}
- _workflowConfig.getTaskDag().validate();
+ _workflowConfig.getJobDag().validate();
- for (String node : _taskConfigs.keySet()) {
+ for (String node : _jobConfigs.keySet()) {
buildConfig(node);
}
}
- /** Builds a TaskConfig from config map. Useful for validating configs */
- private TaskConfig buildConfig(String task) {
- return TaskConfig.Builder.fromMap(_taskConfigs.get(task)).build();
+ /** Builds a JobConfig from config map. Useful for validating configs */
+ private JobConfig buildConfig(String job) {
+ JobConfig.Builder b = JobConfig.Builder.fromMap(_jobConfigs.get(job));
+ if (_taskConfigs != null && _taskConfigs.containsKey(job)) {
+ b.addTaskConfigs(_taskConfigs.get(job));
+ }
+ return b.build();
}
/** Build a workflow incrementally from dependencies and single configs, validate at build time */
public static class Builder {
- private final String _name;
- private final TaskDag _dag;
- private final Map<String, Map<String, String>> _taskConfigs;
+ private String _name;
+ private JobDag _dag;
+ private Map<String, Map<String, String>> _jobConfigs;
+ private Map<String, List<TaskConfig>> _taskConfigs;
private long _expiry;
public Builder(String name) {
_name = name;
- _dag = new TaskDag();
- _taskConfigs = new TreeMap<String, Map<String, String>>();
+ _dag = new JobDag();
+ _jobConfigs = new TreeMap<String, Map<String, String>>();
+ _taskConfigs = new TreeMap<String, List<TaskConfig>>();
_expiry = -1;
}
public Builder addConfig(String node, String key, String val) {
node = namespacify(node);
_dag.addNode(node);
+ if (!_jobConfigs.containsKey(node)) {
+ _jobConfigs.put(node, new TreeMap<String, String>());
+ }
+ _jobConfigs.get(node).put(key, val);
+ return this;
+ }
- if (!_taskConfigs.containsKey(node)) {
- _taskConfigs.put(node, new TreeMap<String, String>());
+ public Builder addJobConfigMap(String node, Map<String, String> jobConfigMap) {
+ return addConfig(node, JobConfig.JOB_CONFIG_MAP, TaskUtil.serializeJobConfigMap(jobConfigMap));
+ }
+
+ public Builder addJobConfig(String node, JobConfig jobConfig) {
+ for (Map.Entry<String, String> e : jobConfig.getResourceConfigMap().entrySet()) {
+ String key = e.getKey();
+ String val = e.getValue();
+ addConfig(node, key, val);
}
- _taskConfigs.get(node).put(key, val);
+ addTaskConfigs(node, jobConfig.getTaskConfigMap().values());
+ return this;
+ }
+ public Builder addTaskConfigs(String node, Collection<TaskConfig> taskConfigs) {
+ node = namespacify(node);
+ _dag.addNode(node);
+ if (!_taskConfigs.containsKey(node)) {
+ _taskConfigs.put(node, new ArrayList<TaskConfig>());
+ }
+ if (!_jobConfigs.containsKey(node)) {
+ _jobConfigs.put(node, new TreeMap<String, String>());
+ }
+ _taskConfigs.get(node).addAll(taskConfigs);
return this;
}
@@ -226,13 +276,13 @@ public class Workflow {
}
public String namespacify(String task) {
- return TaskUtil.getNamespacedTaskName(_name, task);
+ return TaskUtil.getNamespacedJobName(_name, task);
}
public Workflow build() {
- for (String task : _taskConfigs.keySet()) {
+ for (String task : _jobConfigs.keySet()) {
// addConfig(task, TaskConfig.WORKFLOW_ID, _name);
- _taskConfigs.get(task).put(TaskConfig.WORKFLOW_ID, _name);
+ _jobConfigs.get(task).put(JobConfig.WORKFLOW_ID, _name);
}
WorkflowConfig.Builder builder = new WorkflowConfig.Builder();
@@ -242,7 +292,8 @@ public class Workflow {
builder.setExpiry(_expiry);
}
- return new Workflow(_name, builder.build(), _taskConfigs); // calls validate internally
+ return new Workflow(_name, builder.build(), _jobConfigs, _taskConfigs); // calls validate
+ // internally
}
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/WorkflowConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/WorkflowConfig.java b/helix-core/src/main/java/org/apache/helix/task/WorkflowConfig.java
index bb88be7..6f10955 100644
--- a/helix-core/src/main/java/org/apache/helix/task/WorkflowConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/task/WorkflowConfig.java
@@ -34,18 +34,18 @@ public class WorkflowConfig {
public static final long DEFAULT_EXPIRY = 24 * 60 * 60 * 1000;
/* Member variables */
- private final TaskDag _taskDag;
- private final TargetState _targetState;
- private final long _expiry;
+ private JobDag _jobDag;
+ private TargetState _targetState;
+ private long _expiry;
- private WorkflowConfig(TaskDag taskDag, TargetState targetState, long expiry) {
- _taskDag = taskDag;
+ private WorkflowConfig(JobDag jobDag, TargetState targetState, long expiry) {
+ _jobDag = jobDag;
_targetState = targetState;
_expiry = expiry;
}
- public TaskDag getTaskDag() {
- return _taskDag;
+ public JobDag getJobDag() {
+ return _jobDag;
}
public TargetState getTargetState() {
@@ -57,7 +57,7 @@ public class WorkflowConfig {
}
public static class Builder {
- private TaskDag _taskDag = TaskDag.EMPTY_DAG;
+ private JobDag _taskDag = JobDag.EMPTY_DAG;
private TargetState _targetState = TargetState.START;
private long _expiry = DEFAULT_EXPIRY;
@@ -71,7 +71,7 @@ public class WorkflowConfig {
return new WorkflowConfig(_taskDag, _targetState, _expiry);
}
- public Builder setTaskDag(TaskDag v) {
+ public Builder setTaskDag(JobDag v) {
_taskDag = v;
return this;
}
@@ -93,7 +93,7 @@ public class WorkflowConfig {
b.setExpiry(Long.parseLong(cfg.get(EXPIRY)));
}
if (cfg.containsKey(DAG)) {
- b.setTaskDag(TaskDag.fromJson(cfg.get(DAG)));
+ b.setTaskDag(JobDag.fromJson(cfg.get(DAG)));
}
if (cfg.containsKey(TARGET_STATE)) {
b.setTargetState(TargetState.valueOf(cfg.get(TARGET_STATE)));
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/WorkflowContext.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/WorkflowContext.java b/helix-core/src/main/java/org/apache/helix/task/WorkflowContext.java
index cd30860..4feda1b 100644
--- a/helix-core/src/main/java/org/apache/helix/task/WorkflowContext.java
+++ b/helix-core/src/main/java/org/apache/helix/task/WorkflowContext.java
@@ -19,35 +19,21 @@ package org.apache.helix.task;
* under the License.
*/
-import org.apache.helix.HelixProperty;
-import org.apache.helix.ZNRecord;
-
import java.util.Map;
import java.util.TreeMap;
+import org.apache.helix.HelixProperty;
+import org.apache.helix.ZNRecord;
+
/**
* Typed interface to the workflow context information stored by {@link TaskRebalancer} in the Helix
* property store
*/
public class WorkflowContext extends HelixProperty {
-
- enum WorkflowContextEnum {
- WORKFLOW_STATE("STATE"),
- START_TIME("START_TIME"),
- FINISH_TIME("FINISH_TIME"),
- TASK_STATES("TASK_STATES");
-
- final String _value;
-
- private WorkflowContextEnum(String value) {
- _value = value;
- }
-
- public String value() {
- return _value;
- }
- }
-
+ public static final String WORKFLOW_STATE = "STATE";
+ public static final String START_TIME = "START_TIME";
+ public static final String FINISH_TIME = "FINISH_TIME";
+ public static final String TASK_STATES = "TASK_STATES";
public static final int UNFINISHED = -1;
public WorkflowContext(ZNRecord record) {
@@ -55,18 +41,16 @@ public class WorkflowContext extends HelixProperty {
}
public void setWorkflowState(TaskState s) {
- if (_record.getSimpleField(WorkflowContextEnum.WORKFLOW_STATE.value()) == null) {
- _record.setSimpleField(WorkflowContextEnum.WORKFLOW_STATE.value(), s.name());
- } else if (!_record.getSimpleField(WorkflowContextEnum.WORKFLOW_STATE.value()).equals(
- TaskState.FAILED.name())
- && !_record.getSimpleField(WorkflowContextEnum.WORKFLOW_STATE.value()).equals(
- TaskState.COMPLETED.name())) {
- _record.setSimpleField(WorkflowContextEnum.WORKFLOW_STATE.value(), s.name());
+ if (_record.getSimpleField(WORKFLOW_STATE) == null) {
+ _record.setSimpleField(WORKFLOW_STATE, s.name());
+ } else if (!_record.getSimpleField(WORKFLOW_STATE).equals(TaskState.FAILED.name())
+ && !_record.getSimpleField(WORKFLOW_STATE).equals(TaskState.COMPLETED.name())) {
+ _record.setSimpleField(WORKFLOW_STATE, s.name());
}
}
public TaskState getWorkflowState() {
- String s = _record.getSimpleField(WorkflowContextEnum.WORKFLOW_STATE.value());
+ String s = _record.getSimpleField(WORKFLOW_STATE);
if (s == null) {
return null;
}
@@ -74,22 +58,22 @@ public class WorkflowContext extends HelixProperty {
return TaskState.valueOf(s);
}
- public void setTaskState(String taskResource, TaskState s) {
- Map<String, String> states = _record.getMapField(WorkflowContextEnum.TASK_STATES.value());
+ public void setJobState(String jobResource, TaskState s) {
+ Map<String, String> states = _record.getMapField(TASK_STATES);
if (states == null) {
states = new TreeMap<String, String>();
- _record.setMapField(WorkflowContextEnum.TASK_STATES.value(), states);
+ _record.setMapField(TASK_STATES, states);
}
- states.put(taskResource, s.name());
+ states.put(jobResource, s.name());
}
- public TaskState getTaskState(String taskResource) {
- Map<String, String> states = _record.getMapField(WorkflowContextEnum.TASK_STATES.value());
+ public TaskState getJobState(String jobResource) {
+ Map<String, String> states = _record.getMapField(TASK_STATES);
if (states == null) {
return null;
}
- String s = states.get(taskResource);
+ String s = states.get(jobResource);
if (s == null) {
return null;
}
@@ -98,11 +82,11 @@ public class WorkflowContext extends HelixProperty {
}
public void setStartTime(long t) {
- _record.setSimpleField(WorkflowContextEnum.START_TIME.value(), String.valueOf(t));
+ _record.setSimpleField(START_TIME, String.valueOf(t));
}
public long getStartTime() {
- String tStr = _record.getSimpleField(WorkflowContextEnum.START_TIME.value());
+ String tStr = _record.getSimpleField(START_TIME);
if (tStr == null) {
return -1;
}
@@ -111,11 +95,11 @@ public class WorkflowContext extends HelixProperty {
}
public void setFinishTime(long t) {
- _record.setSimpleField(WorkflowContextEnum.FINISH_TIME.value(), String.valueOf(t));
+ _record.setSimpleField(FINISH_TIME, String.valueOf(t));
}
public long getFinishTime() {
- String tStr = _record.getSimpleField(WorkflowContextEnum.FINISH_TIME.value());
+ String tStr = _record.getSimpleField(FINISH_TIME);
if (tStr == null) {
return UNFINISHED;
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/beans/JobBean.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/beans/JobBean.java b/helix-core/src/main/java/org/apache/helix/task/beans/JobBean.java
new file mode 100644
index 0000000..5e12f19
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/beans/JobBean.java
@@ -0,0 +1,42 @@
+package org.apache.helix.task.beans;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.helix.task.JobConfig;
+
+/**
+ * Bean class used for parsing job definitions from YAML.
+ */
+public class JobBean {
+ public String name;
+ public List<String> parents;
+ public String targetResource;
+ public List<String> targetPartitionStates;
+ public List<String> targetPartitions;
+ public String command;
+ public Map<String, String> jobConfigMap;
+ public List<TaskBean> tasks;
+ public long timeoutPerPartition = JobConfig.DEFAULT_TIMEOUT_PER_TASK;
+ public int numConcurrentTasksPerInstance = JobConfig.DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE;
+ public int maxAttemptsPerPartition = JobConfig.DEFAULT_MAX_ATTEMPTS_PER_TASK;
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/beans/TaskBean.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/beans/TaskBean.java b/helix-core/src/main/java/org/apache/helix/task/beans/TaskBean.java
index 9481c6e..eedccb5 100644
--- a/helix-core/src/main/java/org/apache/helix/task/beans/TaskBean.java
+++ b/helix-core/src/main/java/org/apache/helix/task/beans/TaskBean.java
@@ -19,22 +19,14 @@ package org.apache.helix.task.beans;
* under the License.
*/
-import java.util.List;
import java.util.Map;
-import org.apache.helix.task.TaskConfig;
/**
- * Bean class used for parsing task definitions from YAML.
+ * Describes task-specific configuration, including an arbitrary map of
+ * key-value pairs to pass to the task
*/
+
public class TaskBean {
- public String name;
- public List<String> parents;
- public String targetResource;
- public List<String> targetPartitionStates;
- public List<Integer> targetPartitions;
public String command;
- public Map<String, Object> commandConfig;
- public long timeoutPerPartition = TaskConfig.DEFAULT_TIMEOUT_PER_PARTITION;
- public int numConcurrentTasksPerInstance = TaskConfig.DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE;
- public int maxAttemptsPerPartition = TaskConfig.DEFAULT_MAX_ATTEMPTS_PER_PARTITION;
+ public Map<String, String> taskConfigMap;
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/beans/WorkflowBean.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/beans/WorkflowBean.java b/helix-core/src/main/java/org/apache/helix/task/beans/WorkflowBean.java
index 4e64692..76da4c8 100644
--- a/helix-core/src/main/java/org/apache/helix/task/beans/WorkflowBean.java
+++ b/helix-core/src/main/java/org/apache/helix/task/beans/WorkflowBean.java
@@ -27,5 +27,5 @@ import java.util.List;
public class WorkflowBean {
public String name;
public String expiry;
- public List<TaskBean> tasks;
+ public List<JobBean> jobs;
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
new file mode 100644
index 0000000..1ee3991
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestIndependentTaskRebalancer.java
@@ -0,0 +1,171 @@
+package org.apache.helix.integration.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.TestHelper;
+import org.apache.helix.api.id.StateModelDefId;
+import org.apache.helix.integration.ZkIntegrationTestBase;
+import org.apache.helix.integration.manager.ClusterControllerManager;
+import org.apache.helix.integration.manager.MockParticipantManager;
+import org.apache.helix.integration.task.TestTaskRebalancerStopResume.ReindexTask;
+import org.apache.helix.participant.StateMachineEngine;
+import org.apache.helix.task.JobConfig;
+import org.apache.helix.task.Task;
+import org.apache.helix.task.TaskCallbackContext;
+import org.apache.helix.task.TaskConfig;
+import org.apache.helix.task.TaskDriver;
+import org.apache.helix.task.TaskFactory;
+import org.apache.helix.task.TaskResult;
+import org.apache.helix.task.TaskState;
+import org.apache.helix.task.TaskStateModelFactory;
+import org.apache.helix.task.Workflow;
+import org.apache.helix.tools.ClusterSetup;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import org.testng.collections.Sets;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+public class TestIndependentTaskRebalancer extends ZkIntegrationTestBase {
+ private static final int n = 5;
+ private static final int START_PORT = 12918;
+ private final String CLUSTER_NAME = CLUSTER_PREFIX + "_" + getShortClassName();
+ private final MockParticipantManager[] _participants = new MockParticipantManager[n];
+ private ClusterControllerManager _controller;
+ private Set<String> _invokedClasses = Sets.newHashSet();
+
+ private HelixManager _manager;
+ private TaskDriver _driver;
+
+ @BeforeClass
+ public void beforeClass() throws Exception {
+ String namespace = "/" + CLUSTER_NAME;
+ if (_gZkClient.exists(namespace)) {
+ _gZkClient.deleteRecursive(namespace);
+ }
+
+ // Setup cluster and instances
+ ClusterSetup setupTool = new ClusterSetup(ZK_ADDR);
+ setupTool.addCluster(CLUSTER_NAME, true);
+ for (int i = 0; i < n; i++) {
+ String storageNodeName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
+ setupTool.addInstanceToCluster(CLUSTER_NAME, storageNodeName);
+ }
+
+ // Set task callbacks
+ Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
+ taskFactoryReg.put("TaskOne", new TaskFactory() {
+ @Override
+ public Task createNewTask(TaskCallbackContext context) {
+ return new TaskOne(context);
+ }
+ });
+ taskFactoryReg.put("TaskTwo", new TaskFactory() {
+ @Override
+ public Task createNewTask(TaskCallbackContext context) {
+ return new TaskTwo(context);
+ }
+ });
+
+ // start dummy participants
+ for (int i = 0; i < n; i++) {
+ String instanceName = PARTICIPANT_PREFIX + "_" + (START_PORT + i);
+ _participants[i] = new MockParticipantManager(ZK_ADDR, CLUSTER_NAME, instanceName);
+
+ // Register a Task state model factory.
+ StateMachineEngine stateMachine = _participants[i].getStateMachineEngine();
+ stateMachine.registerStateModelFactory(StateModelDefId.from("Task"),
+ new TaskStateModelFactory(_participants[i], taskFactoryReg));
+ _participants[i].syncStart();
+ }
+
+ // Start controller
+ String controllerName = CONTROLLER_PREFIX + "_0";
+ _controller = new ClusterControllerManager(ZK_ADDR, CLUSTER_NAME, controllerName);
+ _controller.syncStart();
+
+ // Start an admin connection
+ _manager =
+ HelixManagerFactory.getZKHelixManager(CLUSTER_NAME, "Admin", InstanceType.ADMINISTRATOR,
+ ZK_ADDR);
+ _manager.connect();
+ _driver = new TaskDriver(_manager);
+ }
+
+ @BeforeMethod
+ public void beforeMethod() {
+ _invokedClasses.clear();
+ }
+
+ @Test
+ public void testDifferentTasks() throws Exception {
+ // Create a job with two different tasks
+ String jobName = TestHelper.getTestMethodName();
+ Workflow.Builder workflowBuilder = new Workflow.Builder(jobName);
+ List<TaskConfig> taskConfigs = Lists.newArrayListWithCapacity(2);
+ TaskConfig taskConfig1 = new TaskConfig("TaskOne", null);
+ TaskConfig taskConfig2 = new TaskConfig("TaskTwo", null);
+ taskConfigs.add(taskConfig1);
+ taskConfigs.add(taskConfig2);
+ workflowBuilder.addTaskConfigs(jobName, taskConfigs);
+ workflowBuilder.addConfig(jobName, JobConfig.COMMAND, "DummyCommand");
+ Map<String, String> jobConfigMap = Maps.newHashMap();
+ jobConfigMap.put("Timeout", "1000");
+ workflowBuilder.addJobConfigMap(jobName, jobConfigMap);
+ _driver.start(workflowBuilder.build());
+
+ // Ensure the job completes
+ TestUtil.pollForWorkflowState(_manager, jobName, TaskState.IN_PROGRESS);
+ TestUtil.pollForWorkflowState(_manager, jobName, TaskState.COMPLETED);
+
+ // Ensure that each class was invoked
+ Assert.assertTrue(_invokedClasses.contains(TaskOne.class.getName()));
+ Assert.assertTrue(_invokedClasses.contains(TaskTwo.class.getName()));
+ }
+
+ private class TaskOne extends ReindexTask {
+ public TaskOne(TaskCallbackContext context) {
+ super(context);
+ }
+
+ @Override
+ public TaskResult run() {
+ _invokedClasses.add(getClass().getName());
+ return super.run();
+ }
+ }
+
+ private class TaskTwo extends TaskOne {
+ public TaskTwo(TaskCallbackContext context) {
+ super(context);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/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 1c83291..0a59ee1 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
@@ -19,9 +19,9 @@ package org.apache.helix.integration.task;
* under the License.
*/
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import java.util.TreeMap;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixDataAccessor;
@@ -34,10 +34,11 @@ import org.apache.helix.integration.ZkIntegrationTestBase;
import org.apache.helix.integration.manager.ClusterControllerManager;
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.Task;
-import org.apache.helix.task.TaskConfig;
+import org.apache.helix.task.TaskCallbackContext;
import org.apache.helix.task.TaskConstants;
-import org.apache.helix.task.TaskContext;
import org.apache.helix.task.TaskDriver;
import org.apache.helix.task.TaskFactory;
import org.apache.helix.task.TaskPartitionState;
@@ -56,11 +57,13 @@ import org.testng.annotations.Test;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
public class TestTaskRebalancer extends ZkIntegrationTestBase {
private static final int n = 5;
private static final int START_PORT = 12918;
private static final String MASTER_SLAVE_STATE_MODEL = "MasterSlave";
+ private static final String TIMEOUT_CONFIG = "Timeout";
private static final int NUM_PARTITIONS = 20;
private static final int NUM_REPLICAS = 3;
private final String CLUSTER_NAME = CLUSTER_PREFIX + "_" + getShortClassName();
@@ -92,8 +95,8 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
taskFactoryReg.put("Reindex", new TaskFactory() {
@Override
- public Task createNewTask(String config) {
- return new ReindexTask(config);
+ public Task createNewTask(TaskCallbackContext context) {
+ return new ReindexTask(context);
}
});
@@ -152,29 +155,30 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
@Test
public void testExpiry() throws Exception {
- String taskName = "Expiry";
+ String jobName = "Expiry";
long expiry = 1000;
+ Map<String, String> commandConfig = ImmutableMap.of(TIMEOUT_CONFIG, String.valueOf(100));
Workflow flow =
WorkflowGenerator
- .generateDefaultSingleTaskWorkflowBuilderWithExtraConfigs(taskName,
- TaskConfig.COMMAND_CONFIG, String.valueOf(100)).setExpiry(expiry).build();
+ .generateDefaultSingleJobWorkflowBuilderWithExtraConfigs(jobName, commandConfig)
+ .setExpiry(expiry).build();
_driver.start(flow);
- TestUtil.pollForWorkflowState(_manager, taskName, TaskState.IN_PROGRESS);
+ TestUtil.pollForWorkflowState(_manager, jobName, TaskState.IN_PROGRESS);
// Running workflow should have config and context viewable through accessor
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
- PropertyKey workflowCfgKey = accessor.keyBuilder().resourceConfig(taskName);
+ PropertyKey workflowCfgKey = accessor.keyBuilder().resourceConfig(jobName);
String workflowPropStoreKey =
- Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, taskName);
+ Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, jobName);
// Ensure context and config exist
Assert.assertTrue(_manager.getHelixPropertyStore().exists(workflowPropStoreKey,
AccessOption.PERSISTENT));
Assert.assertNotSame(accessor.getProperty(workflowCfgKey), null);
- // Wait for task to finish and expire
- TestUtil.pollForWorkflowState(_manager, taskName, TaskState.COMPLETED);
+ // Wait for job to finish and expire
+ TestUtil.pollForWorkflowState(_manager, jobName, TaskState.COMPLETED);
Thread.sleep(expiry);
_driver.invokeRebalance();
Thread.sleep(expiry);
@@ -185,25 +189,26 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
Assert.assertEquals(accessor.getProperty(workflowCfgKey), null);
}
- private void basic(long taskCompletionTime) throws Exception {
+ private void basic(long jobCompletionTime) throws Exception {
// We use a different resource name in each test method as a work around for a helix participant
// bug where it does
// not clear locally cached state when a resource partition is dropped. Once that is fixed we
// should change these
// tests to use the same resource name and implement a beforeMethod that deletes the task
// resource.
- final String taskResource = "basic" + taskCompletionTime;
+ final String jobResource = "basic" + jobCompletionTime;
+ Map<String, String> commandConfig =
+ ImmutableMap.of(TIMEOUT_CONFIG, String.valueOf(jobCompletionTime));
Workflow flow =
- WorkflowGenerator.generateDefaultSingleTaskWorkflowBuilderWithExtraConfigs(taskResource,
- TaskConfig.COMMAND_CONFIG, String.valueOf(taskCompletionTime)).build();
+ WorkflowGenerator.generateDefaultSingleJobWorkflowBuilderWithExtraConfigs(jobResource,
+ commandConfig).build();
_driver.start(flow);
- // Wait for task completion
- TestUtil.pollForWorkflowState(_manager, taskResource, TaskState.COMPLETED);
+ // Wait for job completion
+ TestUtil.pollForWorkflowState(_manager, jobResource, TaskState.COMPLETED);
// Ensure all partitions are completed individually
- TaskContext ctx =
- TaskUtil.getTaskContext(_manager, TaskUtil.getNamespacedTaskName(taskResource));
+ JobContext ctx = TaskUtil.getJobContext(_manager, TaskUtil.getNamespacedJobName(jobResource));
for (int i = 0; i < NUM_PARTITIONS; i++) {
Assert.assertEquals(ctx.getPartitionState(i), TaskPartitionState.COMPLETED);
Assert.assertEquals(ctx.getPartitionNumAttempts(i), 1);
@@ -212,29 +217,31 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
@Test
public void partitionSet() throws Exception {
- final String taskResource = "partitionSet";
- ImmutableList<Integer> targetPartitions = ImmutableList.of(1, 2, 3, 5, 8, 13);
+ final String jobResource = "partitionSet";
+ ImmutableList<String> targetPartitions =
+ ImmutableList.of("TestDB_1", "TestDB_2", "TestDB_3", "TestDB_5", "TestDB_8", "TestDB_13");
// construct and submit our basic workflow
+ Map<String, String> commandConfig = ImmutableMap.of(TIMEOUT_CONFIG, String.valueOf(100));
Workflow flow =
- WorkflowGenerator.generateDefaultSingleTaskWorkflowBuilderWithExtraConfigs(taskResource,
- TaskConfig.COMMAND_CONFIG, String.valueOf(100), TaskConfig.MAX_ATTEMPTS_PER_PARTITION,
- String.valueOf(1), TaskConfig.TARGET_PARTITIONS, Joiner.on(",").join(targetPartitions))
- .build();
+ WorkflowGenerator.generateDefaultSingleJobWorkflowBuilderWithExtraConfigs(jobResource,
+ commandConfig, JobConfig.MAX_ATTEMPTS_PER_TASK, String.valueOf(1),
+ JobConfig.TARGET_PARTITIONS, Joiner.on(",").join(targetPartitions)).build();
_driver.start(flow);
- // wait for task completeness/timeout
- TestUtil.pollForWorkflowState(_manager, taskResource, TaskState.COMPLETED);
+ // wait for job completeness/timeout
+ TestUtil.pollForWorkflowState(_manager, jobResource, TaskState.COMPLETED);
// see if resulting context completed successfully for our partition set
- String namespacedName = TaskUtil.getNamespacedTaskName(taskResource);
+ String namespacedName = TaskUtil.getNamespacedJobName(jobResource);
- TaskContext ctx = TaskUtil.getTaskContext(_manager, namespacedName);
- WorkflowContext workflowContext = TaskUtil.getWorkflowContext(_manager, taskResource);
+ JobContext ctx = TaskUtil.getJobContext(_manager, namespacedName);
+ WorkflowContext workflowContext = TaskUtil.getWorkflowContext(_manager, jobResource);
Assert.assertNotNull(ctx);
Assert.assertNotNull(workflowContext);
- Assert.assertEquals(workflowContext.getTaskState(namespacedName), TaskState.COMPLETED);
- for (int i : targetPartitions) {
+ Assert.assertEquals(workflowContext.getJobState(namespacedName), TaskState.COMPLETED);
+ for (String pName : targetPartitions) {
+ int i = ctx.getPartitionsByTarget().get(pName).get(0);
Assert.assertEquals(ctx.getPartitionState(i), TaskPartitionState.COMPLETED);
Assert.assertEquals(ctx.getPartitionNumAttempts(i), 1);
}
@@ -244,33 +251,32 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
public void testRepeatedWorkflow() throws Exception {
String workflowName = "SomeWorkflow";
Workflow flow =
- WorkflowGenerator.generateDefaultRepeatedTaskWorkflowBuilder(workflowName).build();
+ WorkflowGenerator.generateDefaultRepeatedJobWorkflowBuilder(workflowName).build();
new TaskDriver(_manager).start(flow);
- // Wait until the task completes
+ // Wait until the workflow completes
TestUtil.pollForWorkflowState(_manager, workflowName, TaskState.COMPLETED);
// Assert completion for all tasks within two minutes
- for (String task : flow.getTaskConfigs().keySet()) {
- TestUtil.pollForTaskState(_manager, workflowName, task, TaskState.COMPLETED);
+ for (String task : flow.getJobConfigs().keySet()) {
+ TestUtil.pollForJobState(_manager, workflowName, task, TaskState.COMPLETED);
}
}
@Test
public void timeouts() throws Exception {
- final String taskResource = "timeouts";
+ final String jobResource = "timeouts";
Workflow flow =
- WorkflowGenerator.generateDefaultSingleTaskWorkflowBuilderWithExtraConfigs(taskResource,
- TaskConfig.MAX_ATTEMPTS_PER_PARTITION, String.valueOf(2),
- TaskConfig.TIMEOUT_PER_PARTITION, String.valueOf(100)).build();
+ WorkflowGenerator.generateDefaultSingleJobWorkflowBuilderWithExtraConfigs(jobResource,
+ WorkflowGenerator.DEFAULT_COMMAND_CONFIG, JobConfig.MAX_ATTEMPTS_PER_TASK,
+ String.valueOf(2), JobConfig.TIMEOUT_PER_TASK, String.valueOf(100)).build();
_driver.start(flow);
- // Wait until the task reports failure.
- TestUtil.pollForWorkflowState(_manager, taskResource, TaskState.FAILED);
+ // Wait until the job reports failure.
+ TestUtil.pollForWorkflowState(_manager, jobResource, TaskState.FAILED);
// Check that all partitions timed out up to maxAttempts
- TaskContext ctx =
- TaskUtil.getTaskContext(_manager, TaskUtil.getNamespacedTaskName(taskResource));
+ JobContext ctx = TaskUtil.getJobContext(_manager, TaskUtil.getNamespacedJobName(jobResource));
int maxAttempts = 0;
for (int i = 0; i < NUM_PARTITIONS; i++) {
TaskPartitionState state = ctx.getPartitionState(i);
@@ -282,57 +288,17 @@ public class TestTaskRebalancer extends ZkIntegrationTestBase {
Assert.assertEquals(maxAttempts, 2);
}
- @Test
- public void testIndependentTask() throws Exception {
- final String taskResource = "independentTask";
- Map<String, String> config = new TreeMap<String, String>();
- config.put("TargetPartitions", "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19");
- config.put("Command", "Reindex");
- config.put("CommandConfig", String.valueOf(200));
- config.put("TimeoutPerPartition", String.valueOf(10 * 1000));
- Workflow flow =
- WorkflowGenerator.generateSingleTaskWorkflowBuilder(taskResource, config).build();
- _driver.start(flow);
-
- // Wait for task completion
- TestUtil.pollForWorkflowState(_manager, taskResource, TaskState.COMPLETED);
-
- // Ensure all partitions are completed individually
- TaskContext ctx =
- TaskUtil.getTaskContext(_manager, TaskUtil.getNamespacedTaskName(taskResource));
- for (int i = 0; i < NUM_PARTITIONS; i++) {
- Assert.assertEquals(ctx.getPartitionState(i), TaskPartitionState.COMPLETED);
- Assert.assertEquals(ctx.getPartitionNumAttempts(i), 1);
- }
- }
-
- @Test
- public void testIndependentRepeatedWorkflow() throws Exception {
- final String workflowName = "independentTaskWorkflow";
- Map<String, String> config = new TreeMap<String, String>();
- config.put("TargetPartitions", "0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19");
- config.put("Command", "Reindex");
- config.put("CommandConfig", String.valueOf(200));
- config.put("TimeoutPerPartition", String.valueOf(10 * 1000));
- Workflow flow =
- WorkflowGenerator.generateRepeatedTaskWorkflowBuilder(workflowName, config).build();
- new TaskDriver(_manager).start(flow);
-
- // Wait until the task completes
- TestUtil.pollForWorkflowState(_manager, workflowName, TaskState.COMPLETED);
-
- // Assert completion for all tasks within two minutes
- for (String task : flow.getTaskConfigs().keySet()) {
- TestUtil.pollForTaskState(_manager, workflowName, task, TaskState.COMPLETED);
- }
- }
-
private static class ReindexTask implements Task {
private final long _delay;
private volatile boolean _canceled;
- public ReindexTask(String cfg) {
- _delay = Long.parseLong(cfg);
+ public ReindexTask(TaskCallbackContext context) {
+ JobConfig jobCfg = context.getJobConfig();
+ Map<String, String> cfg = jobCfg.getJobConfigMap();
+ if (cfg == null) {
+ cfg = Collections.emptyMap();
+ }
+ _delay = cfg.containsKey(TIMEOUT_CONFIG) ? Long.parseLong(cfg.get(TIMEOUT_CONFIG)) : 200L;
}
@Override
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
index bb490ea..e555468 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskRebalancerStopResume.java
@@ -19,6 +19,7 @@ package org.apache.helix.integration.task;
* under the License.
*/
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -30,8 +31,9 @@ import org.apache.helix.integration.ZkIntegrationTestBase;
import org.apache.helix.integration.manager.ClusterControllerManager;
import org.apache.helix.integration.manager.MockParticipantManager;
import org.apache.helix.participant.StateMachineEngine;
+import org.apache.helix.task.JobConfig;
import org.apache.helix.task.Task;
-import org.apache.helix.task.TaskConfig;
+import org.apache.helix.task.TaskCallbackContext;
import org.apache.helix.task.TaskDriver;
import org.apache.helix.task.TaskFactory;
import org.apache.helix.task.TaskResult;
@@ -46,13 +48,16 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import com.google.common.collect.ImmutableMap;
+
public class TestTaskRebalancerStopResume extends ZkIntegrationTestBase {
private static final Logger LOG = Logger.getLogger(TestTaskRebalancerStopResume.class);
private static final int n = 5;
private static final int START_PORT = 12918;
private static final String MASTER_SLAVE_STATE_MODEL = "MasterSlave";
+ private static final String TIMEOUT_CONFIG = "Timeout";
private static final String TGT_DB = "TestDB";
- private static final String TASK_RESOURCE = "SomeTask";
+ private static final String JOB_RESOURCE = "SomeJob";
private static final int NUM_PARTITIONS = 20;
private static final int NUM_REPLICAS = 3;
private final String CLUSTER_NAME = CLUSTER_PREFIX + "_" + getShortClassName();
@@ -83,8 +88,8 @@ public class TestTaskRebalancerStopResume extends ZkIntegrationTestBase {
Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
taskFactoryReg.put("Reindex", new TaskFactory() {
@Override
- public Task createNewTask(String config) {
- return new ReindexTask(config);
+ public Task createNewTask(TaskCallbackContext context) {
+ return new ReindexTask(context);
}
});
@@ -137,27 +142,28 @@ public class TestTaskRebalancerStopResume extends ZkIntegrationTestBase {
@Test
public void stopAndResume() throws Exception {
+ Map<String, String> commandConfig = ImmutableMap.of(TIMEOUT_CONFIG, String.valueOf(100));
Workflow flow =
- WorkflowGenerator.generateDefaultSingleTaskWorkflowBuilderWithExtraConfigs(TASK_RESOURCE,
- TaskConfig.COMMAND_CONFIG, String.valueOf(100)).build();
+ WorkflowGenerator.generateDefaultSingleJobWorkflowBuilderWithExtraConfigs(JOB_RESOURCE,
+ commandConfig).build();
LOG.info("Starting flow " + flow.getName());
_driver.start(flow);
- TestUtil.pollForWorkflowState(_manager, TASK_RESOURCE, TaskState.IN_PROGRESS);
+ TestUtil.pollForWorkflowState(_manager, JOB_RESOURCE, TaskState.IN_PROGRESS);
- LOG.info("Pausing task");
- _driver.stop(TASK_RESOURCE);
- TestUtil.pollForWorkflowState(_manager, TASK_RESOURCE, TaskState.STOPPED);
+ LOG.info("Pausing job");
+ _driver.stop(JOB_RESOURCE);
+ TestUtil.pollForWorkflowState(_manager, JOB_RESOURCE, TaskState.STOPPED);
- LOG.info("Resuming task");
- _driver.resume(TASK_RESOURCE);
- TestUtil.pollForWorkflowState(_manager, TASK_RESOURCE, TaskState.COMPLETED);
+ LOG.info("Resuming job");
+ _driver.resume(JOB_RESOURCE);
+ TestUtil.pollForWorkflowState(_manager, JOB_RESOURCE, TaskState.COMPLETED);
}
@Test
public void stopAndResumeWorkflow() throws Exception {
String workflow = "SomeWorkflow";
- Workflow flow = WorkflowGenerator.generateDefaultRepeatedTaskWorkflowBuilder(workflow).build();
+ Workflow flow = WorkflowGenerator.generateDefaultRepeatedJobWorkflowBuilder(workflow).build();
LOG.info("Starting flow " + workflow);
_driver.start(flow);
@@ -176,8 +182,13 @@ public class TestTaskRebalancerStopResume extends ZkIntegrationTestBase {
private final long _delay;
private volatile boolean _canceled;
- public ReindexTask(String cfg) {
- _delay = Long.parseLong(cfg);
+ public ReindexTask(TaskCallbackContext context) {
+ JobConfig jobCfg = context.getJobConfig();
+ Map<String, String> cfg = jobCfg.getJobConfigMap();
+ if (cfg == null) {
+ cfg = Collections.emptyMap();
+ }
+ _delay = cfg.containsKey(TIMEOUT_CONFIG) ? Long.parseLong(cfg.get(TIMEOUT_CONFIG)) : 200L;
}
@Override
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/test/java/org/apache/helix/integration/task/TestUtil.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestUtil.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestUtil.java
index 2cc6cb8..520d7c0 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestUtil.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestUtil.java
@@ -20,18 +20,17 @@ package org.apache.helix.integration.task;
*/
import org.apache.helix.HelixManager;
-import org.apache.helix.task.*;
-import org.apache.log4j.Logger;
+import org.apache.helix.task.TaskState;
+import org.apache.helix.task.TaskUtil;
+import org.apache.helix.task.WorkflowContext;
import org.testng.Assert;
/**
* Static test utility methods.
*/
public class TestUtil {
- private static final Logger LOG = Logger.getLogger(TestUtil.class);
-
/**
- * Polls {@link org.apache.helix.task.TaskContext} for given task resource until a timeout is
+ * Polls {@link org.apache.helix.task.JobContext} for given task resource until a timeout is
* reached.
* If the task has not reached target state by then, an error is thrown
* @param workflowResource Resource to poll for completeness
@@ -52,15 +51,15 @@ public class TestUtil {
Assert.assertEquals(ctx.getWorkflowState(), state);
}
- public static void pollForTaskState(HelixManager manager, String workflowResource,
- String taskName, TaskState state) throws InterruptedException {
+ public static void pollForJobState(HelixManager manager, String workflowResource,
+ String jobName, TaskState state) throws InterruptedException {
// Wait for completion.
long st = System.currentTimeMillis();
WorkflowContext ctx;
do {
Thread.sleep(100);
ctx = TaskUtil.getWorkflowContext(manager, workflowResource);
- } while ((ctx == null || ctx.getTaskState(taskName) == null || ctx.getTaskState(taskName) != state)
+ } while ((ctx == null || ctx.getJobState(jobName) == null || ctx.getJobState(jobName) != state)
&& System.currentTimeMillis() < st + 2 * 60 * 1000 /* 2 mins */);
Assert.assertNotNull(ctx);
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/test/java/org/apache/helix/integration/task/WorkflowGenerator.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/WorkflowGenerator.java b/helix-core/src/test/java/org/apache/helix/integration/task/WorkflowGenerator.java
index 653d88a..921a5f9 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/WorkflowGenerator.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/WorkflowGenerator.java
@@ -19,72 +19,95 @@ package org.apache.helix.integration.task;
* under the License.
*/
+import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.TreeMap;
+import org.apache.helix.task.JobConfig;
import org.apache.helix.task.Workflow;
+import org.apache.log4j.Logger;
+import org.codehaus.jackson.map.ObjectMapper;
/**
* Convenience class for generating various test workflows
*/
public class WorkflowGenerator {
+ private static final Logger LOG = Logger.getLogger(WorkflowGenerator.class);
+
public static final String DEFAULT_TGT_DB = "TestDB";
- private static final String TASK_NAME_1 = "SomeTask1";
- private static final String TASK_NAME_2 = "SomeTask2";
+ public static final String JOB_NAME_1 = "SomeJob1";
+ public static final String JOB_NAME_2 = "SomeJob2";
- private static final Map<String, String> DEFAULT_TASK_CONFIG;
+ public static final Map<String, String> DEFAULT_JOB_CONFIG;
static {
Map<String, String> tmpMap = new TreeMap<String, String>();
tmpMap.put("TargetResource", DEFAULT_TGT_DB);
tmpMap.put("TargetPartitionStates", "MASTER");
tmpMap.put("Command", "Reindex");
- tmpMap.put("CommandConfig", String.valueOf(2000));
tmpMap.put("TimeoutPerPartition", String.valueOf(10 * 1000));
- DEFAULT_TASK_CONFIG = Collections.unmodifiableMap(tmpMap);
+ DEFAULT_JOB_CONFIG = Collections.unmodifiableMap(tmpMap);
+ }
+
+ public static final Map<String, String> DEFAULT_COMMAND_CONFIG;
+ static {
+ Map<String, String> tmpMap = new TreeMap<String, String>();
+ tmpMap.put("Timeout", String.valueOf(2000));
+ DEFAULT_COMMAND_CONFIG = Collections.unmodifiableMap(tmpMap);
}
- public static Workflow.Builder generateDefaultSingleTaskWorkflowBuilderWithExtraConfigs(
- String taskName, String... cfgs) {
+ public static Workflow.Builder generateDefaultSingleJobWorkflowBuilderWithExtraConfigs(
+ String jobName, Map<String, String> commandConfig, String... cfgs) {
if (cfgs.length % 2 != 0) {
throw new IllegalArgumentException(
"Additional configs should have even number of keys and values");
}
- Workflow.Builder bldr = generateDefaultSingleTaskWorkflowBuilder(taskName);
+ Workflow.Builder bldr = generateDefaultSingleJobWorkflowBuilder(jobName);
for (int i = 0; i < cfgs.length; i += 2) {
- bldr.addConfig(taskName, cfgs[i], cfgs[i + 1]);
+ bldr.addConfig(jobName, cfgs[i], cfgs[i + 1]);
}
return bldr;
}
- public static Workflow.Builder generateDefaultSingleTaskWorkflowBuilder(String taskName) {
- return generateSingleTaskWorkflowBuilder(taskName, DEFAULT_TASK_CONFIG);
+ public static Workflow.Builder generateDefaultSingleJobWorkflowBuilder(String jobName) {
+ return generateSingleJobWorkflowBuilder(jobName, DEFAULT_COMMAND_CONFIG, DEFAULT_JOB_CONFIG);
}
- public static Workflow.Builder generateSingleTaskWorkflowBuilder(String taskName,
- Map<String, String> config) {
- Workflow.Builder builder = new Workflow.Builder(taskName);
+ public static Workflow.Builder generateSingleJobWorkflowBuilder(String jobName,
+ Map<String, String> commandConfig, Map<String, String> config) {
+ Workflow.Builder builder = new Workflow.Builder(jobName);
for (String key : config.keySet()) {
- builder.addConfig(taskName, key, config.get(key));
+ builder.addConfig(jobName, key, config.get(key));
+ }
+ if (commandConfig != null) {
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ String serializedMap = mapper.writeValueAsString(commandConfig);
+ builder.addConfig(jobName, JobConfig.JOB_CONFIG_MAP, serializedMap);
+ } catch (IOException e) {
+ LOG.error("Error serializing " + commandConfig, e);
+ }
}
return builder;
}
- public static Workflow.Builder generateDefaultRepeatedTaskWorkflowBuilder(String workflowName) {
- return generateRepeatedTaskWorkflowBuilder(workflowName, DEFAULT_TASK_CONFIG);
- }
-
- public static Workflow.Builder generateRepeatedTaskWorkflowBuilder(String workflowName,
- Map<String, String> config) {
+ public static Workflow.Builder generateDefaultRepeatedJobWorkflowBuilder(String workflowName) {
Workflow.Builder builder = new Workflow.Builder(workflowName);
- builder.addParentChildDependency(TASK_NAME_1, TASK_NAME_2);
+ builder.addParentChildDependency(JOB_NAME_1, JOB_NAME_2);
- for (String key : config.keySet()) {
- builder.addConfig(TASK_NAME_1, key, config.get(key));
- builder.addConfig(TASK_NAME_2, key, config.get(key));
+ for (String key : DEFAULT_JOB_CONFIG.keySet()) {
+ builder.addConfig(JOB_NAME_1, key, DEFAULT_JOB_CONFIG.get(key));
+ builder.addConfig(JOB_NAME_2, key, DEFAULT_JOB_CONFIG.get(key));
+ }
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ String serializedMap = mapper.writeValueAsString(DEFAULT_COMMAND_CONFIG);
+ builder.addConfig(JOB_NAME_1, JobConfig.JOB_CONFIG_MAP, serializedMap);
+ builder.addConfig(JOB_NAME_2, JobConfig.JOB_CONFIG_MAP, serializedMap);
+ } catch (IOException e) {
+ LOG.error("Error serializing " + DEFAULT_COMMAND_CONFIG, e);
}
-
return builder;
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-provisioning/src/main/java/org/apache/helix/provisioning/tools/TaskManager.java
----------------------------------------------------------------------
diff --git a/helix-provisioning/src/main/java/org/apache/helix/provisioning/tools/TaskManager.java b/helix-provisioning/src/main/java/org/apache/helix/provisioning/tools/TaskManager.java
deleted file mode 100644
index 437880e..0000000
--- a/helix-provisioning/src/main/java/org/apache/helix/provisioning/tools/TaskManager.java
+++ /dev/null
@@ -1,247 +0,0 @@
-package org.apache.helix.provisioning.tools;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-
-import org.I0Itec.zkclient.DataUpdater;
-import org.apache.helix.AccessOption;
-import org.apache.helix.ClusterMessagingService;
-import org.apache.helix.HelixConnection;
-import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.HelixManager;
-import org.apache.helix.HelixRole;
-import org.apache.helix.InstanceType;
-import org.apache.helix.PropertyKey;
-import org.apache.helix.ZNRecord;
-import org.apache.helix.api.State;
-import org.apache.helix.api.id.ClusterId;
-import org.apache.helix.api.id.Id;
-import org.apache.helix.api.id.ParticipantId;
-import org.apache.helix.api.id.PartitionId;
-import org.apache.helix.api.id.ResourceId;
-import org.apache.helix.api.id.SessionId;
-import org.apache.helix.manager.zk.HelixConnectionAdaptor;
-import org.apache.helix.model.ExternalView;
-import org.apache.helix.model.IdealState.IdealStateProperty;
-import org.apache.helix.model.LiveInstance;
-import org.apache.helix.model.ResourceConfiguration;
-import org.apache.helix.task.TaskConfig;
-import org.apache.helix.task.TaskDriver;
-import org.apache.helix.task.TaskPartitionState;
-import org.apache.helix.task.TaskUtil;
-import org.apache.helix.task.Workflow;
-import org.apache.log4j.Logger;
-
-import com.google.common.collect.Maps;
-
-public class TaskManager {
- private static final Logger LOG = Logger.getLogger(TaskManager.class);
-
- private final ClusterId _clusterId;
- private final HelixConnection _connection;
- private final HelixManager _manager;
- private final TaskDriver _driver;
-
- public TaskManager(final ClusterId clusterId, final HelixConnection connection) {
- HelixRole dummyRole = new HelixRole() {
- @Override
- public HelixConnection getConnection() {
- return connection;
- }
-
- @Override
- public ClusterId getClusterId() {
- return clusterId;
- }
-
- @Override
- public Id getId() {
- return clusterId;
- }
-
- @Override
- public InstanceType getType() {
- return InstanceType.ADMINISTRATOR;
- }
-
- @Override
- public ClusterMessagingService getMessagingService() {
- return null;
- }
- };
- _manager = new HelixConnectionAdaptor(dummyRole);
- _driver = new TaskDriver(_manager);
- _clusterId = clusterId;
- _connection = connection;
- }
-
- public boolean createTaskQueue(String queueName, boolean isParallel) {
- Workflow.Builder builder = new Workflow.Builder(queueName);
- builder.addConfig(queueName, TaskConfig.COMMAND, queueName);
- builder.addConfig(queueName, TaskConfig.TARGET_PARTITIONS, "");
- builder.addConfig(queueName, TaskConfig.COMMAND_CONFIG, "");
- builder.addConfig(queueName, TaskConfig.LONG_LIVED + "", String.valueOf(true));
- if (isParallel) {
- builder.addConfig(queueName, TaskConfig.NUM_CONCURRENT_TASKS_PER_INSTANCE,
- String.valueOf(Integer.MAX_VALUE));
- }
- Workflow workflow = builder.build();
- try {
- _driver.start(workflow);
- } catch (Exception e) {
- LOG.error("Failed to start queue " + queueName, e);
- return false;
- }
- return true;
- }
-
- public void addTaskToQueue(final String taskName, final String queueName) {
- // Update the resource config with the new partition count
- HelixDataAccessor accessor = _connection.createDataAccessor(_clusterId);
- PropertyKey.Builder keyBuilder = accessor.keyBuilder();
- final ResourceId resourceId = resourceId(queueName);
- final int[] numPartitions = {
- 0
- };
- DataUpdater<ZNRecord> dataUpdater = new DataUpdater<ZNRecord>() {
- @Override
- public ZNRecord update(ZNRecord currentData) {
- // Update the partition integers to add one to the end, and have that integer map to the
- // task name
- String current = currentData.getSimpleField(TaskConfig.TARGET_PARTITIONS);
- int currentId = 0;
- if (current == null || current.isEmpty()) {
- currentData.setSimpleField(TaskConfig.TARGET_PARTITIONS, String.valueOf(currentId));
- } else {
- String[] parts = current.split(",");
- currentId = parts.length;
- numPartitions[0] = currentId + 1;
- currentData.setSimpleField(TaskConfig.TARGET_PARTITIONS, current + "," + currentId);
- }
- Map<String, String> partitionMap = currentData.getMapField(TaskConfig.TASK_NAME_MAP);
- if (partitionMap == null) {
- partitionMap = Maps.newHashMap();
- currentData.setMapField(TaskConfig.TASK_NAME_MAP, partitionMap);
- }
- partitionMap.put(resourceId.toString() + '_' + currentId, taskName);
- return currentData;
- }
- };
- String configPath = keyBuilder.resourceConfig(resourceId.toString()).getPath();
- List<DataUpdater<ZNRecord>> dataUpdaters = new ArrayList<DataUpdater<ZNRecord>>();
- dataUpdaters.add(dataUpdater);
- accessor.updateChildren(Arrays.asList(configPath), dataUpdaters, AccessOption.PERSISTENT);
-
- // Update the ideal state with the proper partition count
- DataUpdater<ZNRecord> idealStateUpdater = new DataUpdater<ZNRecord>() {
- @Override
- public ZNRecord update(ZNRecord currentData) {
- currentData.setSimpleField(IdealStateProperty.NUM_PARTITIONS.toString(),
- String.valueOf(numPartitions[0]));
- return currentData;
- }
- };
- String idealStatePath = keyBuilder.idealStates(queueName + "_" + queueName).getPath();
- dataUpdaters.clear();
- dataUpdaters.add(idealStateUpdater);
- accessor.updateChildren(Arrays.asList(idealStatePath), dataUpdaters, AccessOption.PERSISTENT);
- }
-
- public void cancelTask(String queueName, String taskName) {
- // Get the mapped task name
- final ResourceId resourceId = resourceId(queueName);
- HelixDataAccessor accessor = _connection.createDataAccessor(_clusterId);
- PropertyKey.Builder keyBuilder = accessor.keyBuilder();
- ResourceConfiguration resourceConfig =
- accessor.getProperty(keyBuilder.resourceConfig(resourceId.stringify()));
- if (resourceConfig == null) {
- LOG.error("Queue " + queueName + " does not exist!");
- return;
- }
- Map<String, String> taskMap = resourceConfig.getRecord().getMapField(TaskConfig.TASK_NAME_MAP);
- if (taskMap == null) {
- LOG.error("Task " + taskName + " in queue " + queueName + " does not exist!");
- return;
- }
- String partitionName = null;
- for (Map.Entry<String, String> e : taskMap.entrySet()) {
- String possiblePartition = e.getKey();
- String possibleTask = e.getValue();
- if (taskName.equals(possibleTask)) {
- partitionName = possiblePartition;
- break;
- }
- }
- if (partitionName == null) {
- LOG.error("Task " + taskName + " in queue " + queueName + " does not exist!");
- return;
- }
-
- // Now search the external view for who is running the task
- ExternalView externalView =
- accessor.getProperty(keyBuilder.externalView(resourceId.toString()));
- if (externalView == null) {
- LOG.error("Queue " + queueName + " was never started!");
- return;
- }
- PartitionId partitionId = PartitionId.from(partitionName);
- Map<ParticipantId, State> stateMap = externalView.getStateMap(partitionId);
- if (stateMap == null || stateMap.isEmpty()) {
- LOG.warn("Task " + taskName + " in queue " + queueName + " is not currently running");
- return;
- }
- ParticipantId targetParticipant = null;
- for (ParticipantId participantId : stateMap.keySet()) {
- targetParticipant = participantId;
- }
- if (targetParticipant == null) {
- LOG.warn("Task " + taskName + " in queue " + queueName + " is not currently running");
- return;
- }
-
- // Send a request to stop to the appropriate live instance
- LiveInstance liveInstance =
- accessor.getProperty(keyBuilder.liveInstance(targetParticipant.toString()));
- if (liveInstance == null) {
- LOG.error("Task " + taskName + " in queue " + queueName
- + " is assigned to a non-running participant");
- return;
- }
- SessionId sessionId = liveInstance.getTypedSessionId();
- TaskUtil.setRequestedState(accessor, targetParticipant.toString(), sessionId.toString(),
- resourceId.toString(), partitionId.toString(), TaskPartitionState.STOPPED);
- LOG.info("Task" + taskName + " for queue " + queueName + " instructed to stop");
- }
-
- public void shutdownQueue(String queueName) {
- // Check if tasks are complete, then set task and workflows to complete
-
- // Otherwise, send a stop for everybody
- _driver.stop(resourceId(queueName).toString());
- }
-
- private ResourceId resourceId(String queueName) {
- return ResourceId.from(queueName + '_' + queueName);
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-provisioning/src/test/java/org/apache/helix/provisioning/tools/TestTaskManager.java
----------------------------------------------------------------------
diff --git a/helix-provisioning/src/test/java/org/apache/helix/provisioning/tools/TestTaskManager.java b/helix-provisioning/src/test/java/org/apache/helix/provisioning/tools/TestTaskManager.java
deleted file mode 100644
index 7d46cff..0000000
--- a/helix-provisioning/src/test/java/org/apache/helix/provisioning/tools/TestTaskManager.java
+++ /dev/null
@@ -1,149 +0,0 @@
-package org.apache.helix.provisioning.tools;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.helix.HelixConnection;
-import org.apache.helix.TestHelper;
-import org.apache.helix.ZkUnitTestBase;
-import org.apache.helix.api.id.ClusterId;
-import org.apache.helix.api.id.StateModelDefId;
-import org.apache.helix.integration.TestHelixConnection;
-import org.apache.helix.integration.manager.ClusterControllerManager;
-import org.apache.helix.integration.manager.MockParticipantManager;
-import org.apache.helix.manager.zk.ZkHelixConnection;
-import org.apache.helix.model.IdealState.RebalanceMode;
-import org.apache.helix.task.Task;
-import org.apache.helix.task.TaskFactory;
-import org.apache.helix.task.TaskResult;
-import org.apache.helix.task.TaskStateModelFactory;
-import org.testng.annotations.Test;
-
-public class TestTaskManager extends ZkUnitTestBase {
- @Test
- public void testBasic() throws Exception {
- final int NUM_PARTICIPANTS = 3;
- final int NUM_PARTITIONS = 1;
- final int NUM_REPLICAS = 1;
-
- String className = TestHelper.getTestClassName();
- String methodName = TestHelper.getTestMethodName();
- String clusterName = className + "_" + methodName;
-
- // Set up cluster
- TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
- "localhost", // participant name prefix
- "TestService", // resource name prefix
- 1, // resources
- NUM_PARTITIONS, // partitions per resource
- NUM_PARTICIPANTS, // number of nodes
- NUM_REPLICAS, // replicas
- "StatelessService", RebalanceMode.FULL_AUTO, // just get everything up
- true); // do rebalance
-
- Map<String, TaskFactory> taskFactoryReg = new HashMap<String, TaskFactory>();
- taskFactoryReg.put("mytask1", new TaskFactory() {
- @Override
- public Task createNewTask(String config) {
- return new MyTask(1);
- }
- });
- taskFactoryReg.put("mytask2", new TaskFactory() {
- @Override
- public Task createNewTask(String config) {
- return new MyTask(2);
- }
- });
- MockParticipantManager[] participants = new MockParticipantManager[NUM_PARTICIPANTS];
- for (int i = 0; i < participants.length; i++) {
- String instanceName = "localhost_" + (12918 + i);
- participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
- participants[i].getStateMachineEngine()
- .registerStateModelFactory(StateModelDefId.from("StatelessService"),
- new TestHelixConnection.MockStateModelFactory());
- participants[i].getStateMachineEngine().registerStateModelFactory(
- StateModelDefId.from("Task"), new TaskStateModelFactory(participants[i], taskFactoryReg));
- participants[i].syncStart();
- }
-
- ClusterControllerManager controller =
- new ClusterControllerManager(ZK_ADDR, clusterName, "controller_1");
- controller.syncStart();
-
- HelixConnection connection = new ZkHelixConnection(ZK_ADDR);
- connection.connect();
- ClusterId clusterId = ClusterId.from(clusterName);
- TaskManager taskManager = new TaskManager(clusterId, connection);
- taskManager.createTaskQueue("myqueue", true);
- taskManager.addTaskToQueue("mytask1", "myqueue");
- Thread.sleep(5000);
- taskManager.addTaskToQueue("mytask2", "myqueue");
- taskManager.cancelTask("myqueue", "mytask1");
-
- controller.syncStop();
- for (MockParticipantManager participant : participants) {
- participant.syncStop();
- }
- }
-
- public static class MyTask implements Task {
- private final int _id;
- private Thread _t;
- private TaskResult.Status _status = null;
-
- public MyTask(int id) {
- _id = id;
- }
-
- @Override
- public TaskResult run() {
- _t = new Thread() {
- @Override
- public void run() {
- try {
- Thread.sleep(60000);
- _status = TaskResult.Status.COMPLETED;
- System.err.println("task complete for " + _id);
- } catch (InterruptedException e) {
- _status = TaskResult.Status.CANCELED;
- System.err.println("task canceled for " + _id);
- interrupt();
- }
- }
- };
- _t.start();
- try {
- _t.join();
- } catch (InterruptedException e) {
- _status = TaskResult.Status.CANCELED;
- }
- return new TaskResult(_status, "");
- }
-
- @Override
- public void cancel() {
- if (_t != null && _t.isAlive()) {
- _t.interrupt();
- }
- }
- }
-}
[3/3] git commit: Port recent task framework changes
Posted by ka...@apache.org.
Port recent task framework changes
Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/97ca4de4
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/97ca4de4
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/97ca4de4
Branch: refs/heads/helix-provisioning
Commit: 97ca4de4af522ec8dd8ab2e63b30a34d2a455d34
Parents: 8f0b7e4
Author: Kanak Biscuitwala <ka...@apache.org>
Authored: Wed Apr 30 11:39:23 2014 -0700
Committer: Kanak Biscuitwala <ka...@apache.org>
Committed: Wed Apr 30 11:39:23 2014 -0700
----------------------------------------------------------------------
.../handling/HelixStateTransitionHandler.java | 2 +
.../helix/task/AbstractTaskRebalancer.java | 639 -------------------
.../helix/task/FixedTargetTaskRebalancer.java | 162 +++++
.../helix/task/GenericTaskRebalancer.java | 196 ++++++
.../helix/task/IndependentTaskRebalancer.java | 137 ----
.../java/org/apache/helix/task/JobConfig.java | 334 ++++++++++
.../java/org/apache/helix/task/JobContext.java | 227 +++++++
.../main/java/org/apache/helix/task/JobDag.java | 151 +++++
.../java/org/apache/helix/task/TargetState.java | 8 +-
.../apache/helix/task/TaskCallbackContext.java | 67 ++
.../java/org/apache/helix/task/TaskConfig.java | 359 +++--------
.../java/org/apache/helix/task/TaskContext.java | 135 ----
.../java/org/apache/helix/task/TaskDag.java | 152 -----
.../java/org/apache/helix/task/TaskDriver.java | 132 ++--
.../java/org/apache/helix/task/TaskFactory.java | 5 +-
.../org/apache/helix/task/TaskRebalancer.java | 625 ++++++++++++++++--
.../java/org/apache/helix/task/TaskRunner.java | 1 -
.../org/apache/helix/task/TaskStateModel.java | 43 +-
.../java/org/apache/helix/task/TaskUtil.java | 133 ++--
.../java/org/apache/helix/task/Workflow.java | 175 +++--
.../org/apache/helix/task/WorkflowConfig.java | 20 +-
.../org/apache/helix/task/WorkflowContext.java | 64 +-
.../org/apache/helix/task/beans/JobBean.java | 42 ++
.../org/apache/helix/task/beans/TaskBean.java | 16 +-
.../apache/helix/task/beans/WorkflowBean.java | 2 +-
.../task/TestIndependentTaskRebalancer.java | 171 +++++
.../integration/task/TestTaskRebalancer.java | 152 ++---
.../task/TestTaskRebalancerStopResume.java | 43 +-
.../apache/helix/integration/task/TestUtil.java | 15 +-
.../integration/task/WorkflowGenerator.java | 75 ++-
.../helix/provisioning/tools/TaskManager.java | 247 -------
.../provisioning/tools/TestTaskManager.java | 149 -----
32 files changed, 2478 insertions(+), 2201 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
index 55d8965..1bb6506 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/HelixStateTransitionHandler.java
@@ -58,6 +58,8 @@ import org.apache.log4j.Logger;
public class HelixStateTransitionHandler extends MessageHandler {
public static class HelixStateMismatchException extends Exception {
+ private static final long serialVersionUID = -7669959598697794766L;
+
public HelixStateMismatchException(String info) {
super(info);
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/AbstractTaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskRebalancer.java b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskRebalancer.java
deleted file mode 100644
index f733fb5..0000000
--- a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskRebalancer.java
+++ /dev/null
@@ -1,639 +0,0 @@
-package org.apache.helix.task;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.SortedSet;
-import java.util.TreeMap;
-import java.util.TreeSet;
-
-import org.apache.helix.AccessOption;
-import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.HelixManager;
-import org.apache.helix.PropertyKey;
-import org.apache.helix.ZNRecord;
-import org.apache.helix.api.Cluster;
-import org.apache.helix.api.State;
-import org.apache.helix.api.accessor.ResourceAccessor;
-import org.apache.helix.api.id.ParticipantId;
-import org.apache.helix.api.id.PartitionId;
-import org.apache.helix.api.id.ResourceId;
-import org.apache.helix.controller.context.ControllerContextProvider;
-import org.apache.helix.controller.rebalancer.HelixRebalancer;
-import org.apache.helix.controller.rebalancer.config.BasicRebalancerConfig;
-import org.apache.helix.controller.rebalancer.config.PartitionedRebalancerConfig;
-import org.apache.helix.controller.rebalancer.config.RebalancerConfig;
-import org.apache.helix.controller.stages.ResourceCurrentState;
-import org.apache.helix.model.IdealState;
-import org.apache.helix.model.ResourceAssignment;
-import org.apache.log4j.Logger;
-
-import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Sets;
-
-/**
- * Custom rebalancer implementation for the {@code Task} state model. Abstract task rebalancer with
- * a pluggable assignment policy.
- */
-public abstract class AbstractTaskRebalancer implements HelixRebalancer {
- private static final Logger LOG = Logger.getLogger(AbstractTaskRebalancer.class);
- private HelixManager _manager;
-
- /**
- * Get all the partitions that should be created by this task
- * @param taskCfg the task configuration
- * @param workflowCfg the workflow configuration
- * @param workflowCtx the workflow context
- * @param cluster cluster snapshot
- * @return set of partition numbers
- */
- public abstract Set<Integer> getAllTaskPartitions(TaskConfig taskCfg, WorkflowConfig workflowCfg,
- WorkflowContext workflowCtx, Cluster cluster);
-
- /**
- * Compute an assignment of tasks to instances
- * @param currStateOutput the current state of the instances
- * @param prevAssignment the previous task partition assignment
- * @param instanceList the instances
- * @param taskCfg the task configuration
- * @param taskCtx the task context
- * @param workflowCfg the workflow configuration
- * @param workflowCtx the workflow context
- * @param partitionSet the partitions to assign
- * @param cluster cluster snapshot
- * @return map of instances to set of partition numbers
- */
- public abstract Map<String, SortedSet<Integer>> getTaskAssignment(
- ResourceCurrentState currStateOutput, ResourceAssignment prevAssignment,
- Iterable<ParticipantId> instanceList, TaskConfig taskCfg, TaskContext taskContext,
- WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Set<Integer> partitionSet,
- Cluster cluster);
-
- @Override
- public void init(HelixManager helixManager, ControllerContextProvider contextProvider) {
- _manager = helixManager;
- }
-
- @Override
- public ResourceAssignment computeResourceMapping(RebalancerConfig rebalancerConfig,
- ResourceAssignment helixPrevAssignment, Cluster cluster, ResourceCurrentState currentState) {
- final ResourceId resourceId = rebalancerConfig.getResourceId();
- final String resourceName = resourceId.stringify();
-
- // Fetch task configuration
- TaskConfig taskCfg = TaskUtil.getTaskCfg(_manager, resourceName);
- String workflowResource = taskCfg.getWorkflow();
-
- // Fetch workflow configuration and context
- WorkflowConfig workflowCfg = TaskUtil.getWorkflowCfg(_manager, workflowResource);
- WorkflowContext workflowCtx = TaskUtil.getWorkflowContext(_manager, workflowResource);
-
- // Initialize workflow context if needed
- if (workflowCtx == null) {
- workflowCtx = new WorkflowContext(new ZNRecord("WorkflowContext"));
- workflowCtx.setStartTime(System.currentTimeMillis());
- }
-
- // Check parent dependencies
- for (String parent : workflowCfg.getTaskDag().getDirectParents(resourceName)) {
- if (workflowCtx.getTaskState(parent) == null
- || !workflowCtx.getTaskState(parent).equals(TaskState.COMPLETED)) {
- return emptyAssignment(resourceId);
- }
- }
-
- // Clean up if workflow marked for deletion
- TargetState targetState = workflowCfg.getTargetState();
- if (targetState == TargetState.DELETE) {
- cleanup(_manager, resourceName, workflowCfg, workflowResource);
- return emptyAssignment(resourceId);
- }
-
- // Check if this workflow has been finished past its expiry.
- if (workflowCtx.getFinishTime() != WorkflowContext.UNFINISHED
- && workflowCtx.getFinishTime() + workflowCfg.getExpiry() <= System.currentTimeMillis()) {
- markForDeletion(_manager, workflowResource);
- cleanup(_manager, resourceName, workflowCfg, workflowResource);
- return emptyAssignment(resourceId);
- }
-
- // Fetch any existing context information from the property store.
- TaskContext taskCtx = TaskUtil.getTaskContext(_manager, resourceName);
- if (taskCtx == null) {
- taskCtx = new TaskContext(new ZNRecord("TaskContext"));
- taskCtx.setStartTime(System.currentTimeMillis());
- }
-
- // The task is already in a final state (completed/failed).
- if (workflowCtx.getTaskState(resourceName) == TaskState.FAILED
- || workflowCtx.getTaskState(resourceName) == TaskState.COMPLETED) {
- return emptyAssignment(resourceId);
- }
-
- // Will contain the list of partitions that must be explicitly dropped from the ideal state that
- // is stored in zk.
- // Fetch the previous resource assignment from the property store. This is required because of
- // HELIX-230.
- Set<Integer> partitionsToDrop = new TreeSet<Integer>();
- Set<ParticipantId> liveInstances = cluster.getLiveParticipantMap().keySet();
- ResourceAssignment prevAssignment = TaskUtil.getPrevResourceAssignment(_manager, resourceName);
- if (prevAssignment == null) {
- prevAssignment = new ResourceAssignment(resourceId);
- }
- ResourceAssignment newAssignment =
- computeResourceMapping(resourceName, workflowCfg, taskCfg, prevAssignment, liveInstances,
- currentState, workflowCtx, taskCtx, partitionsToDrop, cluster);
-
- PartitionedRebalancerConfig userConfig =
- BasicRebalancerConfig.convert(rebalancerConfig, PartitionedRebalancerConfig.class);
- if (!partitionsToDrop.isEmpty()) {
- for (Integer pId : partitionsToDrop) {
- userConfig.getPartitionMap().remove(PartitionId.from(pName(resourceName, pId)));
- }
- HelixDataAccessor accessor = _manager.getHelixDataAccessor();
- PropertyKey propertyKey = accessor.keyBuilder().idealStates(resourceName);
-
- IdealState taskIs =
- ResourceAccessor.rebalancerConfigToIdealState(rebalancerConfig,
- cluster.getResource(resourceId).getBucketSize(), cluster.getResource(resourceId)
- .getBatchMessageMode());
- accessor.setProperty(propertyKey, taskIs);
- }
-
- // Update rebalancer context, previous ideal state.
- TaskUtil.setTaskContext(_manager, resourceName, taskCtx);
- TaskUtil.setWorkflowContext(_manager, workflowResource, workflowCtx);
- TaskUtil.setPrevResourceAssignment(_manager, resourceName, newAssignment);
-
- return newAssignment;
- }
-
- private ResourceAssignment computeResourceMapping(String taskResource,
- WorkflowConfig workflowConfig, TaskConfig taskCfg, ResourceAssignment prevAssignment,
- Iterable<ParticipantId> liveInstances, ResourceCurrentState currStateOutput,
- WorkflowContext workflowCtx, TaskContext taskCtx, Set<Integer> partitionsToDropFromIs,
- Cluster cluster) {
- TargetState taskTgtState = workflowConfig.getTargetState();
-
- // Update running status in workflow context
- if (taskTgtState == TargetState.STOP) {
- workflowCtx.setTaskState(taskResource, TaskState.STOPPED);
- // Workflow has been stopped if all tasks are stopped
- if (isWorkflowStopped(workflowCtx, workflowConfig)) {
- workflowCtx.setWorkflowState(TaskState.STOPPED);
- }
- } else {
- workflowCtx.setTaskState(taskResource, TaskState.IN_PROGRESS);
- // Workflow is in progress if any task is in progress
- workflowCtx.setWorkflowState(TaskState.IN_PROGRESS);
- }
-
- // Used to keep track of task partitions that have already been assigned to instances.
- Set<Integer> assignedPartitions = new HashSet<Integer>();
-
- // Keeps a mapping of (partition) -> (instance, state)
- Map<Integer, PartitionAssignment> paMap = new TreeMap<Integer, PartitionAssignment>();
-
- // Process all the current assignments of task partitions.
- Set<Integer> allPartitions =
- getAllTaskPartitions(taskCfg, workflowConfig, workflowCtx, cluster);
- Map<String, SortedSet<Integer>> taskAssignments =
- getTaskPartitionAssignments(liveInstances, prevAssignment, allPartitions);
- for (String instance : taskAssignments.keySet()) {
- Set<Integer> pSet = taskAssignments.get(instance);
- // Used to keep track of partitions that are in one of the final states: COMPLETED, TIMED_OUT,
- // TASK_ERROR, ERROR.
- Set<Integer> donePartitions = new TreeSet<Integer>();
- for (int pId : pSet) {
- final String pName = pName(taskResource, pId);
-
- // Check for pending state transitions on this (partition, instance).
- State s =
- currStateOutput.getPendingState(ResourceId.from(taskResource), PartitionId.from(pName),
- ParticipantId.from(instance));
- String pendingState = (s == null ? null : s.toString());
- if (pendingState != null) {
- // There is a pending state transition for this (partition, instance). Just copy forward
- // the state assignment from the previous ideal state.
- Map<ParticipantId, State> stateMap =
- prevAssignment.getReplicaMap(PartitionId.from(pName));
- if (stateMap != null) {
- State prevState = stateMap.get(ParticipantId.from(instance));
- paMap.put(pId, new PartitionAssignment(instance, prevState.toString()));
- assignedPartitions.add(pId);
- if (LOG.isDebugEnabled()) {
- LOG.debug(String
- .format("Task partition %s has a pending state transition on instance %s."
- + " Using the previous ideal state which was %s.", pName, instance, prevState));
- }
- }
-
- continue;
- }
-
- // Current state is either present or dropped
- State currentState =
- currStateOutput.getCurrentState(ResourceId.from(taskResource), PartitionId.from(pName),
- ParticipantId.from(instance));
- String currentStateStr =
- currentState != null ? currentState.toString() : TaskPartitionState.DROPPED.toString();
- TaskPartitionState currState = TaskPartitionState.valueOf(currentStateStr);
-
- // Process any requested state transitions.
- State reqS =
- currStateOutput.getRequestedState(ResourceId.from(taskResource),
- PartitionId.from(pName), ParticipantId.from(instance));
- String requestedStateStr = (reqS == null ? null : reqS.toString());
- if (requestedStateStr != null && !requestedStateStr.isEmpty()) {
- TaskPartitionState requestedState = TaskPartitionState.valueOf(requestedStateStr);
- if (requestedState.equals(currState)) {
- LOG.warn(String.format(
- "Requested state %s is the same as the current state for instance %s.",
- requestedState, instance));
- }
-
- paMap.put(pId, new PartitionAssignment(instance, requestedState.name()));
- assignedPartitions.add(pId);
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format(
- "Instance %s requested a state transition to %s for partition %s.", instance,
- requestedState, pName));
- }
- continue;
- }
-
- switch (currState) {
- case RUNNING:
- case STOPPED: {
- TaskPartitionState nextState;
- if (taskTgtState == TargetState.START) {
- nextState = TaskPartitionState.RUNNING;
- } else {
- nextState = TaskPartitionState.STOPPED;
- }
-
- paMap.put(pId, new PartitionAssignment(instance, nextState.name()));
- assignedPartitions.add(pId);
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format("Setting task partition %s state to %s on instance %s.", pName,
- nextState, instance));
- }
- }
- break;
- case COMPLETED: {
- // The task has completed on this partition. Mark as such in the context object.
- donePartitions.add(pId);
- if (LOG.isDebugEnabled()) {
- LOG.debug(String
- .format(
- "Task partition %s has completed with state %s. Marking as such in rebalancer context.",
- pName, currState));
- }
- partitionsToDropFromIs.add(pId);
- markPartitionCompleted(taskCtx, pId);
- }
- break;
- case TIMED_OUT:
- case TASK_ERROR:
- case ERROR: {
- donePartitions.add(pId); // The task may be rescheduled on a different instance.
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format(
- "Task partition %s has error state %s. Marking as such in rebalancer context.",
- pName, currState));
- }
- markPartitionError(taskCtx, pId, currState);
- // The error policy is to fail the task as soon a single partition fails for a specified
- // maximum number of attempts.
- if (taskCtx.getPartitionNumAttempts(pId) >= taskCfg.getMaxAttemptsPerPartition()) {
- workflowCtx.setTaskState(taskResource, TaskState.FAILED);
- workflowCtx.setWorkflowState(TaskState.FAILED);
- partitionsToDropFromIs.addAll(allPartitions);
- return emptyAssignment(ResourceId.from(taskResource));
- }
- }
- break;
- case INIT:
- case DROPPED: {
- // currState in [INIT, DROPPED]. Do nothing, the partition is eligible to be reassigned.
- donePartitions.add(pId);
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format(
- "Task partition %s has state %s. It will be dropped from the current ideal state.",
- pName, currState));
- }
- }
- break;
- default:
- throw new AssertionError("Unknown enum symbol: " + currState);
- }
- }
-
- // Remove the set of task partitions that are completed or in one of the error states.
- pSet.removeAll(donePartitions);
- }
-
- if (isTaskComplete(taskCtx, allPartitions)) {
- if (!taskCfg.isLongLived()) {
- workflowCtx.setTaskState(taskResource, TaskState.COMPLETED);
- }
- if (isWorkflowComplete(workflowCtx, workflowConfig)) {
- workflowCtx.setWorkflowState(TaskState.COMPLETED);
- workflowCtx.setFinishTime(System.currentTimeMillis());
- }
- }
-
- // Make additional task assignments if needed.
- if (taskTgtState == TargetState.START) {
- // Contains the set of task partitions that must be excluded from consideration when making
- // any new assignments.
- // This includes all completed, failed, already assigned partitions.
- Set<Integer> excludeSet = Sets.newTreeSet(assignedPartitions);
- addCompletedPartitions(excludeSet, taskCtx, allPartitions);
- // Get instance->[partition, ...] mappings for the target resource.
- Map<String, SortedSet<Integer>> tgtPartitionAssignments =
- getTaskAssignment(currStateOutput, prevAssignment, liveInstances, taskCfg, taskCtx,
- workflowConfig, workflowCtx, allPartitions, cluster);
- for (Map.Entry<String, SortedSet<Integer>> entry : taskAssignments.entrySet()) {
- String instance = entry.getKey();
- // Contains the set of task partitions currently assigned to the instance.
- Set<Integer> pSet = entry.getValue();
- int numToAssign = taskCfg.getNumConcurrentTasksPerInstance() - pSet.size();
- if (numToAssign > 0) {
- List<Integer> nextPartitions =
- getNextPartitions(tgtPartitionAssignments.get(instance), excludeSet, numToAssign);
- for (Integer pId : nextPartitions) {
- String pName = pName(taskResource, pId);
- paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.RUNNING.name()));
- excludeSet.add(pId);
- if (LOG.isDebugEnabled()) {
- LOG.debug(String.format("Setting task partition %s state to %s on instance %s.",
- pName, TaskPartitionState.RUNNING, instance));
- }
- }
- }
- }
- }
-
- // Construct a ResourceAssignment object from the map of partition assignments.
- ResourceAssignment ra = new ResourceAssignment(ResourceId.from(taskResource));
- for (Map.Entry<Integer, PartitionAssignment> e : paMap.entrySet()) {
- PartitionAssignment pa = e.getValue();
- ra.addReplicaMap(PartitionId.from(pName(taskResource, e.getKey())),
- ImmutableMap.of(ParticipantId.from(pa._instance), State.from(pa._state)));
- }
-
- return ra;
- }
-
- /**
- * Checks if the task has completed.
- * @param ctx The rebalancer context.
- * @param allPartitions The set of partitions to check.
- * @return true if all task partitions have been marked with status
- * {@link TaskPartitionState#COMPLETED} in the rebalancer
- * context, false otherwise.
- */
- private static boolean isTaskComplete(TaskContext ctx, Set<Integer> allPartitions) {
- for (Integer pId : allPartitions) {
- TaskPartitionState state = ctx.getPartitionState(pId);
- if (state != TaskPartitionState.COMPLETED) {
- return false;
- }
- }
- return true;
- }
-
- /**
- * Checks if the workflow has completed.
- * @param ctx Workflow context containing task states
- * @param cfg Workflow config containing set of tasks
- * @return returns true if all tasks are {@link TaskState#COMPLETED}, false otherwise.
- */
- private static boolean isWorkflowComplete(WorkflowContext ctx, WorkflowConfig cfg) {
- for (String task : cfg.getTaskDag().getAllNodes()) {
- if (ctx.getTaskState(task) != TaskState.COMPLETED) {
- return false;
- }
- }
- return true;
- }
-
- /**
- * Checks if the workflow has been stopped.
- * @param ctx Workflow context containing task states
- * @param cfg Workflow config containing set of tasks
- * @return returns true if all tasks are {@link TaskState#STOPPED}, false otherwise.
- */
- private static boolean isWorkflowStopped(WorkflowContext ctx, WorkflowConfig cfg) {
- for (String task : cfg.getTaskDag().getAllNodes()) {
- if (ctx.getTaskState(task) != TaskState.STOPPED && ctx.getTaskState(task) != null) {
- return false;
- }
- }
- return true;
- }
-
- private static void markForDeletion(HelixManager mgr, String resourceName) {
- mgr.getConfigAccessor().set(
- TaskUtil.getResourceConfigScope(mgr.getClusterName(), resourceName),
- WorkflowConfig.TARGET_STATE, TargetState.DELETE.name());
- }
-
- /**
- * Cleans up all Helix state associated with this task, wiping workflow-level information if this
- * is the last remaining task in its workflow.
- */
- private static void cleanup(HelixManager mgr, String resourceName, WorkflowConfig cfg,
- String workflowResource) {
- HelixDataAccessor accessor = mgr.getHelixDataAccessor();
- // Delete resource configs.
- PropertyKey cfgKey = getConfigPropertyKey(accessor, resourceName);
- if (!accessor.removeProperty(cfgKey)) {
- throw new RuntimeException(String.format("Error occurred while trying to clean up task %s. "
- + "Failed to remove node %s from Helix. Aborting further clean up steps.", resourceName,
- cfgKey));
- }
- // Delete property store information for this resource.
- String propStoreKey = getRebalancerPropStoreKey(resourceName);
- if (!mgr.getHelixPropertyStore().remove(propStoreKey, AccessOption.PERSISTENT)) {
- throw new RuntimeException(String.format("Error occurred while trying to clean up task %s. "
- + "Failed to remove node %s from Helix. Aborting further clean up steps.", resourceName,
- propStoreKey));
- }
- // Finally, delete the ideal state itself.
- PropertyKey isKey = getISPropertyKey(accessor, resourceName);
- if (!accessor.removeProperty(isKey)) {
- throw new RuntimeException(String.format(
- "Error occurred while trying to clean up task %s. Failed to remove node %s from Helix.",
- resourceName, isKey));
- }
- LOG.info(String.format("Successfully cleaned up task resource %s.", resourceName));
-
- boolean lastInWorkflow = true;
- for (String task : cfg.getTaskDag().getAllNodes()) {
- // check if property store information or resource configs exist for this task
- if (mgr.getHelixPropertyStore().exists(getRebalancerPropStoreKey(task),
- AccessOption.PERSISTENT)
- || accessor.getProperty(getConfigPropertyKey(accessor, task)) != null
- || accessor.getProperty(getISPropertyKey(accessor, task)) != null) {
- lastInWorkflow = false;
- }
- }
-
- // clean up task-level info if this was the last in workflow
- if (lastInWorkflow) {
- // delete workflow config
- PropertyKey workflowCfgKey = getConfigPropertyKey(accessor, workflowResource);
- if (!accessor.removeProperty(workflowCfgKey)) {
- throw new RuntimeException(String.format(
- "Error occurred while trying to clean up workflow %s. "
- + "Failed to remove node %s from Helix. Aborting further clean up steps.",
- workflowResource, workflowCfgKey));
- }
- // Delete property store information for this workflow
- String workflowPropStoreKey = getRebalancerPropStoreKey(workflowResource);
- if (!mgr.getHelixPropertyStore().remove(workflowPropStoreKey, AccessOption.PERSISTENT)) {
- throw new RuntimeException(String.format(
- "Error occurred while trying to clean up workflow %s. "
- + "Failed to remove node %s from Helix. Aborting further clean up steps.",
- workflowResource, workflowPropStoreKey));
- }
- }
-
- }
-
- private static String getRebalancerPropStoreKey(String resource) {
- return Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resource);
- }
-
- private static PropertyKey getISPropertyKey(HelixDataAccessor accessor, String resource) {
- return accessor.keyBuilder().idealStates(resource);
- }
-
- private static PropertyKey getConfigPropertyKey(HelixDataAccessor accessor, String resource) {
- return accessor.keyBuilder().resourceConfig(resource);
- }
-
- private static ResourceAssignment emptyAssignment(ResourceId resourceId) {
- return new ResourceAssignment(resourceId);
- }
-
- private static void addCompletedPartitions(Set<Integer> set, TaskContext ctx,
- Iterable<Integer> pIds) {
- for (Integer pId : pIds) {
- TaskPartitionState state = ctx.getPartitionState(pId);
- if (state == TaskPartitionState.COMPLETED) {
- set.add(pId);
- }
- }
- }
-
- private static List<Integer> getNextPartitions(SortedSet<Integer> candidatePartitions,
- Set<Integer> excluded, int n) {
- List<Integer> result = new ArrayList<Integer>();
- if (candidatePartitions == null || candidatePartitions.isEmpty()) {
- return result;
- }
- for (Integer pId : candidatePartitions) {
- if (result.size() >= n) {
- break;
- }
-
- if (!excluded.contains(pId)) {
- result.add(pId);
- }
- }
-
- return result;
- }
-
- private static void markPartitionCompleted(TaskContext ctx, int pId) {
- ctx.setPartitionState(pId, TaskPartitionState.COMPLETED);
- ctx.setPartitionFinishTime(pId, System.currentTimeMillis());
- ctx.incrementNumAttempts(pId);
- }
-
- private static void markPartitionError(TaskContext ctx, int pId, TaskPartitionState state) {
- ctx.setPartitionState(pId, state);
- ctx.setPartitionFinishTime(pId, System.currentTimeMillis());
- ctx.incrementNumAttempts(pId);
- }
-
- /**
- * Return the assignment of task partitions per instance.
- */
- private static Map<String, SortedSet<Integer>> getTaskPartitionAssignments(
- Iterable<ParticipantId> instanceList, ResourceAssignment assignment, Set<Integer> includeSet) {
- Map<String, SortedSet<Integer>> result = new HashMap<String, SortedSet<Integer>>();
- for (ParticipantId instance : instanceList) {
- result.put(instance.stringify(), new TreeSet<Integer>());
- }
-
- for (PartitionId partitionId : assignment.getMappedPartitionIds()) {
- int pId = pId(partitionId.stringify());
- if (includeSet.contains(pId)) {
- Map<ParticipantId, State> replicaMap = assignment.getReplicaMap(partitionId);
- for (ParticipantId instance : replicaMap.keySet()) {
- SortedSet<Integer> pList = result.get(instance.stringify());
- if (pList != null) {
- pList.add(pId);
- }
- }
- }
- }
-
- return result;
- }
-
- /**
- * Computes the partition name given the resource name and partition id.
- */
- protected static String pName(String resource, int pId) {
- return resource + "_" + pId;
- }
-
- /**
- * Extracts the partition id from the given partition name.
- */
- protected static int pId(String pName) {
- String[] tokens = pName.split("_");
- return Integer.valueOf(tokens[tokens.length - 1]);
- }
-
- /**
- * An (instance, state) pair.
- */
- private static class PartitionAssignment {
- private final String _instance;
- private final String _state;
-
- private PartitionAssignment(String instance, String state) {
- _instance = instance;
- _state = state;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/FixedTargetTaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/FixedTargetTaskRebalancer.java b/helix-core/src/main/java/org/apache/helix/task/FixedTargetTaskRebalancer.java
new file mode 100644
index 0000000..d1329ee
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/FixedTargetTaskRebalancer.java
@@ -0,0 +1,162 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.apache.helix.api.Cluster;
+import org.apache.helix.api.Resource;
+import org.apache.helix.api.State;
+import org.apache.helix.api.id.ParticipantId;
+import org.apache.helix.api.id.PartitionId;
+import org.apache.helix.api.id.ResourceId;
+import org.apache.helix.controller.stages.ResourceCurrentState;
+import org.apache.helix.model.IdealState;
+import org.apache.helix.model.ResourceAssignment;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+
+/**
+ * A rebalancer for when a task group must be assigned according to partitions/states on a target
+ * resource. Here, tasks are colocated according to where a resource's partitions are, as well as
+ * (if desired) only where those partitions are in a given state.
+ */
+public class FixedTargetTaskRebalancer extends TaskRebalancer {
+
+ @Override
+ public Set<Integer> getAllTaskPartitions(JobConfig jobCfg, JobContext jobCtx,
+ WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Cluster cache) {
+ return getAllTaskPartitions(getTgtIdealState(jobCfg, cache), jobCfg, jobCtx);
+ }
+
+ @Override
+ public Map<ParticipantId, SortedSet<Integer>> getTaskAssignment(
+ ResourceCurrentState currStateOutput, ResourceAssignment prevAssignment,
+ Iterable<ParticipantId> instanceList, JobConfig jobCfg, JobContext jobContext,
+ WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Set<Integer> partitionSet,
+ Cluster cache) {
+ IdealState tgtIs = getTgtIdealState(jobCfg, cache);
+ if (tgtIs == null) {
+ return Collections.emptyMap();
+ }
+ Set<String> tgtStates = jobCfg.getTargetPartitionStates();
+ return getTgtPartitionAssignment(currStateOutput, instanceList, tgtIs, tgtStates, partitionSet,
+ jobContext);
+ }
+
+ /**
+ * Gets the ideal state of the target resource of this job
+ * @param jobCfg job config containing target resource id
+ * @param cluster snapshot of the cluster containing the task and target resource
+ * @return target resource ideal state, or null
+ */
+ private static IdealState getTgtIdealState(JobConfig jobCfg, Cluster cache) {
+ String tgtResourceId = jobCfg.getTargetResource();
+ Resource resource = cache.getResource(ResourceId.from(tgtResourceId));
+ return resource.getIdealState();
+ }
+
+ /**
+ * Returns the set of all partition ids for a job.
+ * <p/>
+ * If a set of partition ids was explicitly specified in the config, that is used. Otherwise, we
+ * use the list of all partition ids from the target resource.
+ */
+ private static Set<Integer> getAllTaskPartitions(IdealState tgtResourceIs, JobConfig jobCfg,
+ JobContext taskCtx) {
+ if (tgtResourceIs == null) {
+ return null;
+ }
+ Map<String, List<Integer>> currentTargets = taskCtx.getPartitionsByTarget();
+ SortedSet<String> targetPartitions = Sets.newTreeSet();
+ if (jobCfg.getTargetPartitions() != null) {
+ targetPartitions.addAll(jobCfg.getTargetPartitions());
+ } else {
+ targetPartitions.addAll(tgtResourceIs.getPartitionSet());
+ }
+
+ Set<Integer> taskPartitions = Sets.newTreeSet();
+ for (String pName : targetPartitions) {
+ taskPartitions.addAll(getPartitionsForTargetPartition(pName, currentTargets, taskCtx));
+ }
+ return taskPartitions;
+ }
+
+ private static List<Integer> getPartitionsForTargetPartition(String targetPartition,
+ Map<String, List<Integer>> currentTargets, JobContext jobCtx) {
+ if (!currentTargets.containsKey(targetPartition)) {
+ int nextId = jobCtx.getPartitionSet().size();
+ jobCtx.setPartitionTarget(nextId, targetPartition);
+ return Lists.newArrayList(nextId);
+ } else {
+ return currentTargets.get(targetPartition);
+ }
+ }
+
+ /**
+ * Get partition assignments for the target resource, but only for the partitions of interest.
+ * @param currStateOutput The current state of the instances in the cluster.
+ * @param instanceList The set of instances.
+ * @param tgtIs The ideal state of the target resource.
+ * @param tgtStates Only partitions in this set of states will be considered. If null, partitions
+ * do not need to
+ * be in any specific state to be considered.
+ * @param includeSet The set of partitions to consider.
+ * @return A map of instance vs set of partition ids assigned to that instance.
+ */
+ private static Map<ParticipantId, SortedSet<Integer>> getTgtPartitionAssignment(
+ ResourceCurrentState currStateOutput, Iterable<ParticipantId> instanceList, IdealState tgtIs,
+ Set<String> tgtStates, Set<Integer> includeSet, JobContext jobCtx) {
+ Map<ParticipantId, SortedSet<Integer>> result =
+ new HashMap<ParticipantId, SortedSet<Integer>>();
+ for (ParticipantId instance : instanceList) {
+ result.put(instance, new TreeSet<Integer>());
+ }
+
+ Map<String, List<Integer>> partitionsByTarget = jobCtx.getPartitionsByTarget();
+ for (String pName : tgtIs.getPartitionSet()) {
+ List<Integer> partitions = partitionsByTarget.get(pName);
+ if (partitions == null || partitions.size() < 1) {
+ continue;
+ }
+ int pId = partitions.get(0);
+ if (includeSet.contains(pId)) {
+ for (ParticipantId instance : instanceList) {
+ State s =
+ currStateOutput.getCurrentState(ResourceId.from(tgtIs.getResourceName()),
+ PartitionId.from(pName), instance);
+ String state = (s == null ? null : s.toString());
+ if (tgtStates == null || tgtStates.contains(state)) {
+ result.get(instance).add(pId);
+ }
+ }
+ }
+ }
+
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/GenericTaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/GenericTaskRebalancer.java b/helix-core/src/main/java/org/apache/helix/task/GenericTaskRebalancer.java
new file mode 100644
index 0000000..8b5a258
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/GenericTaskRebalancer.java
@@ -0,0 +1,196 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.apache.helix.ZNRecord;
+import org.apache.helix.api.Cluster;
+import org.apache.helix.api.Resource;
+import org.apache.helix.api.State;
+import org.apache.helix.api.id.ParticipantId;
+import org.apache.helix.api.id.PartitionId;
+import org.apache.helix.api.id.ResourceId;
+import org.apache.helix.controller.stages.ResourceCurrentState;
+import org.apache.helix.controller.strategy.AutoRebalanceStrategy;
+import org.apache.helix.model.IdealState;
+import org.apache.helix.model.ResourceAssignment;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+/**
+ * This class does an assignment based on an automatic rebalancing strategy, rather than requiring
+ * assignment to target partitions and states of another resource
+ */
+public class GenericTaskRebalancer extends TaskRebalancer {
+ @Override
+ public Set<Integer> getAllTaskPartitions(JobConfig jobCfg, JobContext jobCtx,
+ WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Cluster cache) {
+ Map<String, TaskConfig> taskMap = jobCfg.getTaskConfigMap();
+ Map<String, Integer> taskIdMap = jobCtx.getTaskIdPartitionMap();
+ for (TaskConfig taskCfg : taskMap.values()) {
+ String taskId = taskCfg.getId();
+ int nextPartition = jobCtx.getPartitionSet().size();
+ if (!taskIdMap.containsKey(taskId)) {
+ jobCtx.setTaskIdForPartition(nextPartition, taskId);
+ }
+ }
+ return jobCtx.getPartitionSet();
+ }
+
+ @Override
+ public Map<ParticipantId, SortedSet<Integer>> getTaskAssignment(
+ ResourceCurrentState currStateOutput, ResourceAssignment prevAssignment,
+ Iterable<ParticipantId> instanceList, JobConfig jobCfg, final JobContext jobContext,
+ WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Set<Integer> partitionSet,
+ Cluster cache) {
+ // Gather input to the full auto rebalancing algorithm
+ LinkedHashMap<State, Integer> states = new LinkedHashMap<State, Integer>();
+ states.put(State.from("ONLINE"), 1);
+
+ // Only map partitions whose assignment we care about
+ final Set<TaskPartitionState> honoredStates =
+ Sets.newHashSet(TaskPartitionState.INIT, TaskPartitionState.RUNNING,
+ TaskPartitionState.STOPPED);
+ Set<Integer> filteredPartitionSet = Sets.newHashSet();
+ for (Integer p : partitionSet) {
+ TaskPartitionState state = (jobContext == null) ? null : jobContext.getPartitionState(p);
+ if (state == null || honoredStates.contains(state)) {
+ filteredPartitionSet.add(p);
+ }
+ }
+
+ // Transform from partition id to fully qualified partition name
+ List<Integer> partitionNums = Lists.newArrayList(partitionSet);
+ Collections.sort(partitionNums);
+ final ResourceId resourceId = prevAssignment.getResourceId();
+ List<PartitionId> partitions =
+ new ArrayList<PartitionId>(Lists.transform(partitionNums,
+ new Function<Integer, PartitionId>() {
+ @Override
+ public PartitionId apply(Integer partitionNum) {
+ return PartitionId.from(resourceId + "_" + partitionNum);
+ }
+ }));
+
+ // Compute the current assignment
+ Map<PartitionId, Map<ParticipantId, State>> currentMapping = Maps.newHashMap();
+ for (PartitionId partition : currStateOutput.getCurrentStateMappedPartitions(resourceId)) {
+ if (!filteredPartitionSet.contains(pId(partition.toString()))) {
+ // not computing old partitions
+ continue;
+ }
+ Map<ParticipantId, State> allPreviousDecisionMap = Maps.newHashMap();
+ if (prevAssignment != null) {
+ allPreviousDecisionMap.putAll(prevAssignment.getReplicaMap(partition));
+ }
+ allPreviousDecisionMap.putAll(currStateOutput.getCurrentStateMap(resourceId, partition));
+ allPreviousDecisionMap.putAll(currStateOutput.getPendingStateMap(resourceId, partition));
+ currentMapping.put(partition, allPreviousDecisionMap);
+ }
+
+ // Get the assignment keyed on partition
+ AutoRebalanceStrategy strategy =
+ new AutoRebalanceStrategy(resourceId, partitions, states, Integer.MAX_VALUE,
+ new AutoRebalanceStrategy.DefaultPlacementScheme());
+ List<ParticipantId> allNodes =
+ Lists.newArrayList(getEligibleInstances(jobCfg, currStateOutput, instanceList, cache));
+ Collections.sort(allNodes);
+ ZNRecord record = strategy.typedComputePartitionAssignment(allNodes, currentMapping, allNodes);
+ Map<String, List<String>> preferenceLists = record.getListFields();
+
+ // Convert to an assignment keyed on participant
+ Map<ParticipantId, SortedSet<Integer>> taskAssignment = Maps.newHashMap();
+ for (Map.Entry<String, List<String>> e : preferenceLists.entrySet()) {
+ String partitionName = e.getKey();
+ partitionName = String.valueOf(pId(partitionName));
+ List<String> preferenceList = e.getValue();
+ for (String participantName : preferenceList) {
+ ParticipantId participantId = ParticipantId.from(participantName);
+ if (!taskAssignment.containsKey(participantId)) {
+ taskAssignment.put(participantId, new TreeSet<Integer>());
+ }
+ taskAssignment.get(participantId).add(Integer.valueOf(partitionName));
+ }
+ }
+ return taskAssignment;
+ }
+
+ /**
+ * Filter a list of instances based on targeted resource policies
+ * @param jobCfg the job configuration
+ * @param currStateOutput the current state of all instances in the cluster
+ * @param instanceList valid instances
+ * @param cache current snapshot of the cluster
+ * @return a set of instances that can be assigned to
+ */
+ private Set<ParticipantId> getEligibleInstances(JobConfig jobCfg,
+ ResourceCurrentState currStateOutput, Iterable<ParticipantId> instanceList, Cluster cache) {
+ // No target resource means any instance is available
+ Set<ParticipantId> allInstances = Sets.newHashSet(instanceList);
+ String targetResource = jobCfg.getTargetResource();
+ if (targetResource == null) {
+ return allInstances;
+ }
+
+ // Bad ideal state means don't assign
+ Resource resource = cache.getResource(ResourceId.from(targetResource));
+ IdealState idealState = (resource != null) ? resource.getIdealState() : null;
+ if (idealState == null) {
+ return Collections.emptySet();
+ }
+
+ // Get the partitions on the target resource to use
+ Set<String> partitions = idealState.getPartitionSet();
+ List<String> targetPartitions = jobCfg.getTargetPartitions();
+ if (targetPartitions != null && !targetPartitions.isEmpty()) {
+ partitions.retainAll(targetPartitions);
+ }
+
+ // Based on state matches, add eligible instances
+ Set<ParticipantId> eligibleInstances = Sets.newHashSet();
+ Set<String> targetStates = jobCfg.getTargetPartitionStates();
+ for (String partition : partitions) {
+ Map<ParticipantId, State> stateMap =
+ currStateOutput.getCurrentStateMap(ResourceId.from(targetResource),
+ PartitionId.from(partition));
+ for (Map.Entry<ParticipantId, State> e : stateMap.entrySet()) {
+ ParticipantId instanceName = e.getKey();
+ State state = e.getValue();
+ if (targetStates == null || targetStates.isEmpty()
+ || targetStates.contains(state.toString())) {
+ eligibleInstances.add(instanceName);
+ }
+ }
+ }
+ allInstances.retainAll(eligibleInstances);
+ return allInstances;
+ }
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/IndependentTaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/IndependentTaskRebalancer.java b/helix-core/src/main/java/org/apache/helix/task/IndependentTaskRebalancer.java
deleted file mode 100644
index 2bc4081..0000000
--- a/helix-core/src/main/java/org/apache/helix/task/IndependentTaskRebalancer.java
+++ /dev/null
@@ -1,137 +0,0 @@
-package org.apache.helix.task;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.SortedSet;
-import java.util.TreeSet;
-
-import org.apache.helix.ZNRecord;
-import org.apache.helix.api.Cluster;
-import org.apache.helix.api.State;
-import org.apache.helix.api.id.ParticipantId;
-import org.apache.helix.api.id.PartitionId;
-import org.apache.helix.api.id.ResourceId;
-import org.apache.helix.controller.stages.ResourceCurrentState;
-import org.apache.helix.controller.strategy.AutoRebalanceStrategy;
-import org.apache.helix.model.ResourceAssignment;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
-/**
- * A task rebalancer that evenly assigns tasks to nodes
- */
-public class IndependentTaskRebalancer extends AbstractTaskRebalancer {
-
- @Override
- public Set<Integer> getAllTaskPartitions(TaskConfig taskCfg, WorkflowConfig workflowCfg,
- WorkflowContext workflowCtx, Cluster cluster) {
- Set<Integer> taskPartitions = new HashSet<Integer>();
- if (taskCfg.getTargetPartitions() != null) {
- for (Integer pId : taskCfg.getTargetPartitions()) {
- taskPartitions.add(pId);
- }
- }
- return taskPartitions;
- }
-
- @Override
- public Map<String, SortedSet<Integer>> getTaskAssignment(ResourceCurrentState currStateOutput,
- ResourceAssignment prevAssignment, Iterable<ParticipantId> instanceList, TaskConfig taskCfg,
- final TaskContext taskContext, WorkflowConfig workflowCfg, WorkflowContext workflowCtx,
- Set<Integer> partitionSet, Cluster cluster) {
- // Gather input to the full auto rebalancing algorithm
- LinkedHashMap<State, Integer> states = new LinkedHashMap<State, Integer>();
- states.put(State.from("ONLINE"), 1);
-
- // Only map partitions whose assignment we care about
- final Set<TaskPartitionState> honoredStates =
- Sets.newHashSet(TaskPartitionState.INIT, TaskPartitionState.RUNNING,
- TaskPartitionState.STOPPED);
- Set<Integer> filteredPartitionSet = Sets.newHashSet();
- for (Integer p : partitionSet) {
- TaskPartitionState state = (taskContext == null) ? null : taskContext.getPartitionState(p);
- if (state == null || honoredStates.contains(state)) {
- filteredPartitionSet.add(p);
- }
- }
-
- // Transform from partition id to fully qualified partition name
- List<Integer> partitionNums = Lists.newArrayList(partitionSet);
- Collections.sort(partitionNums);
- final ResourceId resourceId = prevAssignment.getResourceId();
- List<PartitionId> partitions =
- new ArrayList<PartitionId>(Lists.transform(partitionNums,
- new Function<Integer, PartitionId>() {
- @Override
- public PartitionId apply(Integer partitionNum) {
- return PartitionId.from(resourceId, partitionNum.toString());
- }
- }));
-
- // Compute the current assignment
- Map<PartitionId, Map<ParticipantId, State>> currentMapping = Maps.newHashMap();
- for (PartitionId partitionId : currStateOutput.getCurrentStateMappedPartitions(resourceId)) {
- if (!filteredPartitionSet.contains(pId(partitionId.toString()))) {
- // not computing old partitions
- continue;
- }
- Map<ParticipantId, State> allPreviousDecisionMap = Maps.newHashMap();
- if (prevAssignment != null) {
- allPreviousDecisionMap.putAll(prevAssignment.getReplicaMap(partitionId));
- }
- allPreviousDecisionMap.putAll(currStateOutput.getCurrentStateMap(resourceId, partitionId));
- allPreviousDecisionMap.putAll(currStateOutput.getPendingStateMap(resourceId, partitionId));
- currentMapping.put(partitionId, allPreviousDecisionMap);
- }
-
- // Get the assignment keyed on partition
- AutoRebalanceStrategy strategy =
- new AutoRebalanceStrategy(resourceId, partitions, states, Integer.MAX_VALUE,
- new AutoRebalanceStrategy.DefaultPlacementScheme());
- List<ParticipantId> allNodes = Lists.newArrayList(instanceList);
- ZNRecord record = strategy.typedComputePartitionAssignment(allNodes, currentMapping, allNodes);
- Map<String, List<String>> preferenceLists = record.getListFields();
-
- // Convert to an assignment keyed on participant
- Map<String, SortedSet<Integer>> taskAssignment = Maps.newHashMap();
- for (Map.Entry<String, List<String>> e : preferenceLists.entrySet()) {
- String partitionName = e.getKey();
- partitionName = String.valueOf(pId(partitionName));
- List<String> preferenceList = e.getValue();
- for (String participantName : preferenceList) {
- if (!taskAssignment.containsKey(participantName)) {
- taskAssignment.put(participantName, new TreeSet<Integer>());
- }
- taskAssignment.get(participantName).add(Integer.valueOf(partitionName));
- }
- }
- return taskAssignment;
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/JobConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/JobConfig.java b/helix-core/src/main/java/org/apache/helix/task/JobConfig.java
new file mode 100644
index 0000000..90e3cfc
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/JobConfig.java
@@ -0,0 +1,334 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
+
+/**
+ * Provides a typed interface to job configurations.
+ */
+public class JobConfig {
+ // // Property names ////
+
+ /** The name of the workflow to which the job belongs. */
+ public static final String WORKFLOW_ID = "WorkflowID";
+ /** The assignment strategy of this job */
+ public static final String ASSIGNMENT_STRATEGY = "AssignmentStrategy";
+ /** The name of the target resource. */
+ public static final String TARGET_RESOURCE = "TargetResource";
+ /**
+ * The set of the target partition states. The value must be a comma-separated list of partition
+ * states.
+ */
+ public static final String TARGET_PARTITION_STATES = "TargetPartitionStates";
+ /**
+ * The set of the target partition ids. The value must be a comma-separated list of partition ids.
+ */
+ public static final String TARGET_PARTITIONS = "TargetPartitions";
+ /** The command that is to be run by participants in the case of identical tasks. */
+ public static final String COMMAND = "Command";
+ /** The command configuration to be used by the tasks. */
+ public static final String JOB_CONFIG_MAP = "JobConfig";
+ /** The timeout for a task. */
+ public static final String TIMEOUT_PER_TASK = "TimeoutPerPartition";
+ /** The maximum number of times the task rebalancer may attempt to execute a task. */
+ public static final String MAX_ATTEMPTS_PER_TASK = "MaxAttemptsPerTask";
+ /** The number of concurrent tasks that are allowed to run on an instance. */
+ public static final String NUM_CONCURRENT_TASKS_PER_INSTANCE = "ConcurrentTasksPerInstance";
+
+ /** The individual task configurations, if any **/
+ public static final String TASK_CONFIGS = "TaskConfigs";
+
+ // // Default property values ////
+
+ public static final long DEFAULT_TIMEOUT_PER_TASK = 60 * 60 * 1000; // 1 hr.
+ public static final int DEFAULT_MAX_ATTEMPTS_PER_TASK = 10;
+ public static final int DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE = 1;
+
+ private final String _workflow;
+ private final String _targetResource;
+ private final List<String> _targetPartitions;
+ private final Set<String> _targetPartitionStates;
+ private final String _command;
+ private final Map<String, String> _jobConfigMap;
+ private final long _timeoutPerTask;
+ private final int _numConcurrentTasksPerInstance;
+ private final int _maxAttemptsPerTask;
+ private final Map<String, TaskConfig> _taskConfigMap;
+
+ private JobConfig(String workflow, String targetResource, List<String> targetPartitions,
+ Set<String> targetPartitionStates, String command, Map<String, String> jobConfigMap,
+ long timeoutPerTask, int numConcurrentTasksPerInstance, int maxAttemptsPerTask,
+ Map<String, TaskConfig> taskConfigMap) {
+ _workflow = workflow;
+ _targetResource = targetResource;
+ _targetPartitions = targetPartitions;
+ _targetPartitionStates = targetPartitionStates;
+ _command = command;
+ _jobConfigMap = jobConfigMap;
+ _timeoutPerTask = timeoutPerTask;
+ _numConcurrentTasksPerInstance = numConcurrentTasksPerInstance;
+ _maxAttemptsPerTask = maxAttemptsPerTask;
+ if (taskConfigMap != null) {
+ _taskConfigMap = taskConfigMap;
+ } else {
+ _taskConfigMap = Collections.emptyMap();
+ }
+ }
+
+ public String getWorkflow() {
+ return _workflow == null ? Workflow.UNSPECIFIED : _workflow;
+ }
+
+ public String getTargetResource() {
+ return _targetResource;
+ }
+
+ public List<String> getTargetPartitions() {
+ return _targetPartitions;
+ }
+
+ public Set<String> getTargetPartitionStates() {
+ return _targetPartitionStates;
+ }
+
+ public String getCommand() {
+ return _command;
+ }
+
+ public Map<String, String> getJobConfigMap() {
+ return _jobConfigMap;
+ }
+
+ public long getTimeoutPerTask() {
+ return _timeoutPerTask;
+ }
+
+ public int getNumConcurrentTasksPerInstance() {
+ return _numConcurrentTasksPerInstance;
+ }
+
+ public int getMaxAttemptsPerTask() {
+ return _maxAttemptsPerTask;
+ }
+
+ public Map<String, TaskConfig> getTaskConfigMap() {
+ return _taskConfigMap;
+ }
+
+ public TaskConfig getTaskConfig(String id) {
+ return _taskConfigMap.get(id);
+ }
+
+ public Map<String, String> getResourceConfigMap() {
+ Map<String, String> cfgMap = new HashMap<String, String>();
+ cfgMap.put(JobConfig.WORKFLOW_ID, _workflow);
+ if (_command != null) {
+ cfgMap.put(JobConfig.COMMAND, _command);
+ }
+ if (_jobConfigMap != null) {
+ String serializedConfig = TaskUtil.serializeJobConfigMap(_jobConfigMap);
+ if (serializedConfig != null) {
+ cfgMap.put(JobConfig.JOB_CONFIG_MAP, serializedConfig);
+ }
+ }
+ if (_targetResource != null) {
+ cfgMap.put(JobConfig.TARGET_RESOURCE, _targetResource);
+ }
+ if (_targetPartitionStates != null) {
+ cfgMap.put(JobConfig.TARGET_PARTITION_STATES, Joiner.on(",").join(_targetPartitionStates));
+ }
+ if (_targetPartitions != null) {
+ cfgMap.put(JobConfig.TARGET_PARTITIONS, Joiner.on(",").join(_targetPartitions));
+ }
+ cfgMap.put(JobConfig.TIMEOUT_PER_TASK, "" + _timeoutPerTask);
+ cfgMap.put(JobConfig.MAX_ATTEMPTS_PER_TASK, "" + _maxAttemptsPerTask);
+ return cfgMap;
+ }
+
+ /**
+ * A builder for {@link JobConfig}. Validates the configurations.
+ */
+ public static class Builder {
+ private String _workflow;
+ private String _targetResource;
+ private List<String> _targetPartitions;
+ private Set<String> _targetPartitionStates;
+ private String _command;
+ private Map<String, String> _commandConfig;
+ private Map<String, TaskConfig> _taskConfigMap = Maps.newHashMap();
+ private long _timeoutPerTask = DEFAULT_TIMEOUT_PER_TASK;
+ private int _numConcurrentTasksPerInstance = DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE;
+ private int _maxAttemptsPerTask = DEFAULT_MAX_ATTEMPTS_PER_TASK;
+
+ public JobConfig build() {
+ validate();
+
+ return new JobConfig(_workflow, _targetResource, _targetPartitions, _targetPartitionStates,
+ _command, _commandConfig, _timeoutPerTask, _numConcurrentTasksPerInstance,
+ _maxAttemptsPerTask, _taskConfigMap);
+ }
+
+ /**
+ * Convenience method to build a {@link JobConfig} from a {@code Map<String, String>}.
+ * @param cfg A map of property names to their string representations.
+ * @return A {@link Builder}.
+ */
+ public static Builder fromMap(Map<String, String> cfg) {
+ Builder b = new Builder();
+ if (cfg.containsKey(WORKFLOW_ID)) {
+ b.setWorkflow(cfg.get(WORKFLOW_ID));
+ }
+ if (cfg.containsKey(TARGET_RESOURCE)) {
+ b.setTargetResource(cfg.get(TARGET_RESOURCE));
+ }
+ if (cfg.containsKey(TARGET_PARTITIONS)) {
+ b.setTargetPartitions(csvToStringList(cfg.get(TARGET_PARTITIONS)));
+ }
+ if (cfg.containsKey(TARGET_PARTITION_STATES)) {
+ b.setTargetPartitionStates(new HashSet<String>(Arrays.asList(cfg.get(
+ TARGET_PARTITION_STATES).split(","))));
+ }
+ if (cfg.containsKey(COMMAND)) {
+ b.setCommand(cfg.get(COMMAND));
+ }
+ if (cfg.containsKey(JOB_CONFIG_MAP)) {
+ Map<String, String> commandConfigMap =
+ TaskUtil.deserializeJobConfigMap(cfg.get(JOB_CONFIG_MAP));
+ b.setJobConfigMap(commandConfigMap);
+ }
+ if (cfg.containsKey(TIMEOUT_PER_TASK)) {
+ b.setTimeoutPerTask(Long.parseLong(cfg.get(TIMEOUT_PER_TASK)));
+ }
+ if (cfg.containsKey(NUM_CONCURRENT_TASKS_PER_INSTANCE)) {
+ b.setNumConcurrentTasksPerInstance(Integer.parseInt(cfg
+ .get(NUM_CONCURRENT_TASKS_PER_INSTANCE)));
+ }
+ if (cfg.containsKey(MAX_ATTEMPTS_PER_TASK)) {
+ b.setMaxAttemptsPerTask(Integer.parseInt(cfg.get(MAX_ATTEMPTS_PER_TASK)));
+ }
+ return b;
+ }
+
+ public Builder setWorkflow(String v) {
+ _workflow = v;
+ return this;
+ }
+
+ public Builder setTargetResource(String v) {
+ _targetResource = v;
+ return this;
+ }
+
+ public Builder setTargetPartitions(List<String> v) {
+ _targetPartitions = ImmutableList.copyOf(v);
+ return this;
+ }
+
+ public Builder setTargetPartitionStates(Set<String> v) {
+ _targetPartitionStates = ImmutableSet.copyOf(v);
+ return this;
+ }
+
+ public Builder setCommand(String v) {
+ _command = v;
+ return this;
+ }
+
+ public Builder setJobConfigMap(Map<String, String> v) {
+ _commandConfig = v;
+ return this;
+ }
+
+ public Builder setTimeoutPerTask(long v) {
+ _timeoutPerTask = v;
+ return this;
+ }
+
+ public Builder setNumConcurrentTasksPerInstance(int v) {
+ _numConcurrentTasksPerInstance = v;
+ return this;
+ }
+
+ public Builder setMaxAttemptsPerTask(int v) {
+ _maxAttemptsPerTask = v;
+ return this;
+ }
+
+ public Builder addTaskConfigs(List<TaskConfig> taskConfigs) {
+ if (taskConfigs != null) {
+ for (TaskConfig taskConfig : taskConfigs) {
+ _taskConfigMap.put(taskConfig.getId(), taskConfig);
+ }
+ }
+ return this;
+ }
+
+ public Builder addTaskConfigMap(Map<String, TaskConfig> taskConfigMap) {
+ _taskConfigMap.putAll(taskConfigMap);
+ return this;
+ }
+
+ private void validate() {
+ if (_taskConfigMap.isEmpty() && _targetResource == null) {
+ throw new IllegalArgumentException(String.format("%s cannot be null", TARGET_RESOURCE));
+ }
+ if (_taskConfigMap.isEmpty() && _targetPartitionStates != null
+ && _targetPartitionStates.isEmpty()) {
+ throw new IllegalArgumentException(String.format("%s cannot be an empty set",
+ TARGET_PARTITION_STATES));
+ }
+ if (_taskConfigMap.isEmpty() && _command == null) {
+ throw new IllegalArgumentException(String.format("%s cannot be null", COMMAND));
+ }
+ if (_timeoutPerTask < 0) {
+ throw new IllegalArgumentException(String.format("%s has invalid value %s",
+ TIMEOUT_PER_TASK, _timeoutPerTask));
+ }
+ if (_numConcurrentTasksPerInstance < 1) {
+ throw new IllegalArgumentException(String.format("%s has invalid value %s",
+ NUM_CONCURRENT_TASKS_PER_INSTANCE, _numConcurrentTasksPerInstance));
+ }
+ if (_maxAttemptsPerTask < 1) {
+ throw new IllegalArgumentException(String.format("%s has invalid value %s",
+ MAX_ATTEMPTS_PER_TASK, _maxAttemptsPerTask));
+ }
+ if (_workflow == null) {
+ throw new IllegalArgumentException(String.format("%s cannot be null", WORKFLOW_ID));
+ }
+ }
+
+ private static List<String> csvToStringList(String csv) {
+ String[] vals = csv.split(",");
+ return Arrays.asList(vals);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/JobContext.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/JobContext.java b/helix-core/src/main/java/org/apache/helix/task/JobContext.java
new file mode 100644
index 0000000..7742c67
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/JobContext.java
@@ -0,0 +1,227 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.apache.helix.HelixProperty;
+import org.apache.helix.ZNRecord;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+/**
+ * Provides a typed interface to the context information stored by {@link TaskRebalancer} in the
+ * Helix property store.
+ */
+public class JobContext extends HelixProperty {
+ private enum ContextProperties {
+ START_TIME,
+ STATE,
+ NUM_ATTEMPTS,
+ FINISH_TIME,
+ TARGET,
+ TASK_ID
+ }
+
+ public JobContext(ZNRecord record) {
+ super(record);
+ }
+
+ public void setStartTime(long t) {
+ _record.setSimpleField(ContextProperties.START_TIME.toString(), String.valueOf(t));
+ }
+
+ public long getStartTime() {
+ String tStr = _record.getSimpleField(ContextProperties.START_TIME.toString());
+ if (tStr == null) {
+ return -1;
+ }
+
+ return Long.parseLong(tStr);
+ }
+
+ public void setPartitionState(int p, TaskPartitionState s) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ map = new TreeMap<String, String>();
+ _record.setMapField(pStr, map);
+ }
+ map.put(ContextProperties.STATE.toString(), s.name());
+ }
+
+ public TaskPartitionState getPartitionState(int p) {
+ Map<String, String> map = _record.getMapField(String.valueOf(p));
+ if (map == null) {
+ return null;
+ }
+
+ String str = map.get(ContextProperties.STATE.toString());
+ if (str != null) {
+ return TaskPartitionState.valueOf(str);
+ } else {
+ return null;
+ }
+ }
+
+ public void setPartitionNumAttempts(int p, int n) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ map = new TreeMap<String, String>();
+ _record.setMapField(pStr, map);
+ }
+ map.put(ContextProperties.NUM_ATTEMPTS.toString(), String.valueOf(n));
+ }
+
+ public int incrementNumAttempts(int pId) {
+ int n = this.getPartitionNumAttempts(pId);
+ if (n < 0) {
+ n = 0;
+ }
+ n += 1;
+ this.setPartitionNumAttempts(pId, n);
+ return n;
+ }
+
+ public int getPartitionNumAttempts(int p) {
+ Map<String, String> map = _record.getMapField(String.valueOf(p));
+ if (map == null) {
+ return -1;
+ }
+
+ String nStr = map.get(ContextProperties.NUM_ATTEMPTS.toString());
+ if (nStr == null) {
+ return -1;
+ }
+
+ return Integer.parseInt(nStr);
+ }
+
+ public void setPartitionFinishTime(int p, long t) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ map = new TreeMap<String, String>();
+ _record.setMapField(pStr, map);
+ }
+ map.put(ContextProperties.FINISH_TIME.toString(), String.valueOf(t));
+ }
+
+ public long getPartitionFinishTime(int p) {
+ Map<String, String> map = _record.getMapField(String.valueOf(p));
+ if (map == null) {
+ return -1;
+ }
+
+ String tStr = map.get(ContextProperties.FINISH_TIME.toString());
+ if (tStr == null) {
+ return -1;
+ }
+
+ return Long.parseLong(tStr);
+ }
+
+ public void setPartitionTarget(int p, String targetPName) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ map = new TreeMap<String, String>();
+ _record.setMapField(pStr, map);
+ }
+ map.put(ContextProperties.TARGET.toString(), targetPName);
+ }
+
+ public String getTargetForPartition(int p) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ return null;
+ } else {
+ return map.get(ContextProperties.TARGET.toString());
+ }
+ }
+
+ public Map<String, List<Integer>> getPartitionsByTarget() {
+ Map<String, List<Integer>> result = Maps.newHashMap();
+ for (Map.Entry<String, Map<String, String>> mapField : _record.getMapFields().entrySet()) {
+ Integer pId = Integer.parseInt(mapField.getKey());
+ Map<String, String> map = mapField.getValue();
+ String target = map.get(ContextProperties.TARGET.toString());
+ if (target != null) {
+ List<Integer> partitions;
+ if (!result.containsKey(target)) {
+ partitions = Lists.newArrayList();
+ result.put(target, partitions);
+ } else {
+ partitions = result.get(target);
+ }
+ partitions.add(pId);
+ }
+ }
+ return result;
+ }
+
+ public Set<Integer> getPartitionSet() {
+ Set<Integer> partitions = Sets.newHashSet();
+ for (String pName : _record.getMapFields().keySet()) {
+ partitions.add(Integer.valueOf(pName));
+ }
+ return partitions;
+ }
+
+ public void setTaskIdForPartition(int p, String taskId) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ map = new TreeMap<String, String>();
+ _record.setMapField(pStr, map);
+ }
+ map.put(ContextProperties.TASK_ID.toString(), taskId);
+ }
+
+ public String getTaskIdForPartition(int p) {
+ String pStr = String.valueOf(p);
+ Map<String, String> map = _record.getMapField(pStr);
+ if (map == null) {
+ return null;
+ } else {
+ return map.get(ContextProperties.TASK_ID.toString());
+ }
+ }
+
+ public Map<String, Integer> getTaskIdPartitionMap() {
+ Map<String, Integer> partitionMap = new HashMap<String, Integer>();
+ for (Map.Entry<String, Map<String, String>> mapField : _record.getMapFields().entrySet()) {
+ Integer pId = Integer.parseInt(mapField.getKey());
+ Map<String, String> map = mapField.getValue();
+ if (map.containsKey(ContextProperties.TASK_ID.toString())) {
+ partitionMap.put(map.get(ContextProperties.TASK_ID.toString()), pId);
+ }
+ }
+ return partitionMap;
+ }
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/JobDag.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/JobDag.java b/helix-core/src/main/java/org/apache/helix/task/JobDag.java
new file mode 100644
index 0000000..18a721e
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/JobDag.java
@@ -0,0 +1,151 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+import org.codehaus.jackson.annotate.JsonProperty;
+import org.codehaus.jackson.map.ObjectMapper;
+
+/**
+ * Provides a convenient way to construct, traverse,
+ * and validate a job dependency graph
+ */
+public class JobDag {
+ @JsonProperty("parentsToChildren")
+ private Map<String, Set<String>> _parentsToChildren;
+
+ @JsonProperty("childrenToParents")
+ private Map<String, Set<String>> _childrenToParents;
+
+ @JsonProperty("allNodes")
+ private Set<String> _allNodes;
+
+ public static final JobDag EMPTY_DAG = new JobDag();
+
+ public JobDag() {
+ _parentsToChildren = new TreeMap<String, Set<String>>();
+ _childrenToParents = new TreeMap<String, Set<String>>();
+ _allNodes = new TreeSet<String>();
+ }
+
+ public void addParentToChild(String parent, String child) {
+ if (!_parentsToChildren.containsKey(parent)) {
+ _parentsToChildren.put(parent, new TreeSet<String>());
+ }
+ _parentsToChildren.get(parent).add(child);
+
+ if (!_childrenToParents.containsKey(child)) {
+ _childrenToParents.put(child, new TreeSet<String>());
+ }
+ _childrenToParents.get(child).add(parent);
+
+ _allNodes.add(parent);
+ _allNodes.add(child);
+ }
+
+ public void addNode(String node) {
+ _allNodes.add(node);
+ }
+
+ public Map<String, Set<String>> getParentsToChildren() {
+ return _parentsToChildren;
+ }
+
+ public Map<String, Set<String>> getChildrenToParents() {
+ return _childrenToParents;
+ }
+
+ public Set<String> getAllNodes() {
+ return _allNodes;
+ }
+
+ public Set<String> getDirectChildren(String node) {
+ if (!_parentsToChildren.containsKey(node)) {
+ return new TreeSet<String>();
+ }
+ return _parentsToChildren.get(node);
+ }
+
+ public Set<String> getDirectParents(String node) {
+ if (!_childrenToParents.containsKey(node)) {
+ return new TreeSet<String>();
+ }
+ return _childrenToParents.get(node);
+ }
+
+ public String toJson() throws Exception {
+ return new ObjectMapper().writeValueAsString(this);
+ }
+
+ public static JobDag fromJson(String json) {
+ try {
+ return new ObjectMapper().readValue(json, JobDag.class);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unable to parse json " + json + " into job dag");
+ }
+ }
+
+ /**
+ * Checks that dag contains no cycles and all nodes are reachable.
+ */
+ public void validate() {
+ Set<String> prevIteration = new TreeSet<String>();
+
+ // get all unparented nodes
+ for (String node : _allNodes) {
+ if (getDirectParents(node).isEmpty()) {
+ prevIteration.add(node);
+ }
+ }
+
+ // visit children nodes up to max iteration count, by which point we should have exited
+ // naturally
+ Set<String> allNodesReached = new TreeSet<String>();
+ int iterationCount = 0;
+ int maxIterations = _allNodes.size() + 1;
+
+ while (!prevIteration.isEmpty() && iterationCount < maxIterations) {
+ // construct set of all children reachable from prev iteration
+ Set<String> thisIteration = new TreeSet<String>();
+ for (String node : prevIteration) {
+ thisIteration.addAll(getDirectChildren(node));
+ }
+
+ allNodesReached.addAll(prevIteration);
+ prevIteration = thisIteration;
+ iterationCount++;
+ }
+
+ allNodesReached.addAll(prevIteration);
+
+ if (iterationCount >= maxIterations) {
+ throw new IllegalArgumentException("DAG invalid: cycles detected");
+ }
+
+ if (!allNodesReached.containsAll(_allNodes)) {
+ throw new IllegalArgumentException("DAG invalid: unreachable nodes found. Reachable set is "
+ + allNodesReached);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TargetState.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TargetState.java b/helix-core/src/main/java/org/apache/helix/task/TargetState.java
index 0551d6c..4285e67 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TargetState.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TargetState.java
@@ -20,20 +20,20 @@ package org.apache.helix.task;
*/
/**
- * Enumeration of target states for a task.
+ * Enumeration of target states for a job.
*/
public enum TargetState {
/**
- * Indicates that the rebalancer must start/resume the task.
+ * Indicates that the rebalancer must start/resume the job.
*/
START,
/**
- * Indicates that the rebalancer should stop any running task partitions and cease doing any
+ * Indicates that the rebalancer should stop any running tasks and cease doing any
* further task assignments.
*/
STOP,
/**
- * Indicates that the rebalancer must delete this task.
+ * Indicates that the rebalancer must delete this job.
*/
DELETE
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskCallbackContext.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskCallbackContext.java b/helix-core/src/main/java/org/apache/helix/task/TaskCallbackContext.java
new file mode 100644
index 0000000..124ec12
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskCallbackContext.java
@@ -0,0 +1,67 @@
+package org.apache.helix.task;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.HelixManager;
+
+/**
+ * A wrapper for all information about a task and the job of which it is a part.
+ */
+public class TaskCallbackContext {
+ private HelixManager _manager;
+ private TaskConfig _taskConfig;
+ private JobConfig _jobConfig;
+
+ void setManager(HelixManager manager) {
+ _manager = manager;
+ }
+
+ void setTaskConfig(TaskConfig taskConfig) {
+ _taskConfig = taskConfig;
+ }
+
+ void setJobConfig(JobConfig jobConfig) {
+ _jobConfig = jobConfig;
+ }
+
+ /**
+ * Get an active Helix connection
+ * @return HelixManager instance
+ */
+ public HelixManager getManager() {
+ return _manager;
+ }
+
+ /**
+ * Get task-specific configuration properties
+ * @return TaskConfig instance
+ */
+ public TaskConfig getTaskConfig() {
+ return _taskConfig;
+ }
+
+ /**
+ * Get job-specific configuration properties
+ * @return JobConfig instance
+ */
+ public JobConfig getJobConfig() {
+ return _jobConfig;
+ }
+}
[2/3] Port recent task framework changes
Posted by ka...@apache.org.
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskConfig.java b/helix-core/src/main/java/org/apache/helix/task/TaskConfig.java
index 0287657..547ba48 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskConfig.java
@@ -19,315 +19,108 @@ package org.apache.helix.task;
* under the License.
*/
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
+import java.io.IOException;
import java.util.Map;
-import java.util.Set;
+import java.util.UUID;
-import org.apache.helix.task.Workflow.WorkflowEnum;
+import org.apache.helix.task.beans.TaskBean;
+import org.apache.log4j.Logger;
+import org.codehaus.jackson.map.ObjectMapper;
-import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
/**
- * Provides a typed interface to task configurations.
+ * Configuration for an individual task to be run as part of a job.
*/
public class TaskConfig {
- // // Property names ////
-
- /** The name of the workflow to which the task belongs. */
- public static final String WORKFLOW_ID = "WorkflowID";
- /** The name of the target resource. */
- public static final String TARGET_RESOURCE = "TargetResource";
- /**
- * The set of the target partition states. The value must be a comma-separated list of partition
- * states.
- */
- public static final String TARGET_PARTITION_STATES = "TargetPartitionStates";
- /**
- * The set of the target partition ids. The value must be a comma-separated list of partition ids.
- */
- public static final String TARGET_PARTITIONS = "TargetPartitions";
- /** The command that is to be run by participants. */
- public static final String COMMAND = "Command";
- /** The command configuration to be used by the task partitions. */
- public static final String COMMAND_CONFIG = "CommandConfig";
- /** The timeout for a task partition. */
- public static final String TIMEOUT_PER_PARTITION = "TimeoutPerPartition";
- /** The maximum number of times the task rebalancer may attempt to execute a task partitions. */
- public static final String MAX_ATTEMPTS_PER_PARTITION = "MaxAttemptsPerPartition";
- /** The number of concurrent tasks that are allowed to run on an instance. */
- public static final String NUM_CONCURRENT_TASKS_PER_INSTANCE = "ConcurrentTasksPerInstance";
- /** Support overarching tasks that hang around for a while */
- public static final String LONG_LIVED = "LongLived";
- /** Support giving mapping partition IDs to specific task names **/
- public static final String TASK_NAME_MAP = "TaskNameMap";
-
- // // Default property values ////
-
- public static final long DEFAULT_TIMEOUT_PER_PARTITION = 60 * 60 * 1000; // 1 hr.
- public static final int DEFAULT_MAX_ATTEMPTS_PER_PARTITION = 10;
- public static final int DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE = 1;
-
- private final String _workflow;
- private final String _targetResource;
- private final List<Integer> _targetPartitions;
- private final Set<String> _targetPartitionStates;
- private final String _command;
- private final String _commandConfig;
- private final long _timeoutPerPartition;
- private final int _numConcurrentTasksPerInstance;
- private final int _maxAttemptsPerPartition;
- private final boolean _longLived;
- private final Map<String, String> _taskNameMap;
-
- private TaskConfig(String workflow, String targetResource, List<Integer> targetPartitions,
- Set<String> targetPartitionStates, String command, String commandConfig,
- long timeoutPerPartition, int numConcurrentTasksPerInstance, int maxAttemptsPerPartition,
- boolean longLived, Map<String, String> taskNameMap) {
- _workflow = workflow;
- _targetResource = targetResource;
- _targetPartitions = targetPartitions;
- _targetPartitionStates = targetPartitionStates;
- _command = command;
- _commandConfig = commandConfig;
- _timeoutPerPartition = timeoutPerPartition;
- _numConcurrentTasksPerInstance = numConcurrentTasksPerInstance;
- _maxAttemptsPerPartition = maxAttemptsPerPartition;
- _longLived = longLived;
- _taskNameMap = taskNameMap;
- }
-
- public String getWorkflow() {
- return _workflow == null ? WorkflowEnum.UNSPECIFIED.name() : _workflow;
- }
-
- public String getTargetResource() {
- return _targetResource;
- }
-
- public List<Integer> getTargetPartitions() {
- return _targetPartitions;
- }
-
- public Set<String> getTargetPartitionStates() {
- return _targetPartitionStates;
+ private enum TaskConfigFields {
+ TASK_ID,
+ TASK_COMMAND
}
- public String getCommand() {
- return _command;
- }
+ private static final Logger LOG = Logger.getLogger(TaskConfig.class);
- public String getCommandConfig() {
- return _commandConfig;
- }
+ private final Map<String, String> _configMap;
- public long getTimeoutPerPartition() {
- return _timeoutPerPartition;
+ /**
+ * Instantiate the task config
+ * @param command the command to invoke for the task
+ * @param configMap configuration to be passed as part of the invocation
+ * @param id existing task ID
+ */
+ public TaskConfig(String command, Map<String, String> configMap, String id) {
+ if (configMap == null) {
+ configMap = Maps.newHashMap();
+ }
+ if (id == null) {
+ id = UUID.randomUUID().toString();
+ }
+ configMap.put(TaskConfigFields.TASK_COMMAND.toString(), command);
+ configMap.put(TaskConfigFields.TASK_ID.toString(), id);
+ _configMap = configMap;
}
- public int getNumConcurrentTasksPerInstance() {
- return _numConcurrentTasksPerInstance;
+ /**
+ * Instantiate the task config
+ * @param command the command to invoke for the task
+ * @param configMap configuration to be passed as part of the invocation
+ */
+ public TaskConfig(String command, Map<String, String> configMap) {
+ this(command, configMap, null);
}
- public int getMaxAttemptsPerPartition() {
- return _maxAttemptsPerPartition;
+ /**
+ * Unique identifier for this task
+ * @return UUID as a string
+ */
+ public String getId() {
+ return _configMap.get(TaskConfigFields.TASK_ID.toString());
}
- public boolean isLongLived() {
- return _longLived;
+ /**
+ * Get the command to invoke for this task
+ * @return string command
+ */
+ public String getCommand() {
+ return _configMap.get(TaskConfigFields.TASK_COMMAND.toString());
}
- public Map<String, String> getTaskNameMap() {
- return _taskNameMap;
+ /**
+ * Get the configuration map for this task's command
+ * @return map of configuration key to value
+ */
+ public Map<String, String> getConfigMap() {
+ return _configMap;
}
- public Map<String, String> getResourceConfigMap() {
- Map<String, String> cfgMap = new HashMap<String, String>();
- cfgMap.put(TaskConfig.WORKFLOW_ID, _workflow);
- cfgMap.put(TaskConfig.COMMAND, _command);
- cfgMap.put(TaskConfig.COMMAND_CONFIG, _commandConfig);
- cfgMap.put(TaskConfig.TARGET_RESOURCE, _targetResource);
- if (_targetPartitionStates != null) {
- cfgMap.put(TaskConfig.TARGET_PARTITION_STATES, Joiner.on(",").join(_targetPartitionStates));
- }
- if (_targetPartitions != null) {
- cfgMap.put(TaskConfig.TARGET_PARTITIONS, Joiner.on(",").join(_targetPartitions));
+ @Override
+ public String toString() {
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ return mapper.writeValueAsString(this);
+ } catch (IOException e) {
+ LOG.error("Could not serialize TaskConfig", e);
}
- cfgMap.put(TaskConfig.TIMEOUT_PER_PARTITION, "" + _timeoutPerPartition);
- cfgMap.put(TaskConfig.MAX_ATTEMPTS_PER_PARTITION, "" + _maxAttemptsPerPartition);
- cfgMap.put(TaskConfig.LONG_LIVED + "", String.valueOf(_longLived));
- cfgMap.put(TaskConfig.NUM_CONCURRENT_TASKS_PER_INSTANCE + "",
- String.valueOf(_numConcurrentTasksPerInstance));
- return cfgMap;
+ return super.toString();
}
/**
- * A builder for {@link TaskConfig}. Validates the configurations.
+ * Instantiate a typed configuration from a bean
+ * @param bean plain bean describing the task
+ * @return instantiated TaskConfig
*/
- public static class Builder {
- private String _workflow;
- private String _targetResource;
- private List<Integer> _targetPartitions;
- private Set<String> _targetPartitionStates;
- private String _command;
- private String _commandConfig;
- private long _timeoutPerPartition = DEFAULT_TIMEOUT_PER_PARTITION;
- private int _numConcurrentTasksPerInstance = DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE;
- private int _maxAttemptsPerPartition = DEFAULT_MAX_ATTEMPTS_PER_PARTITION;
- private boolean _longLived = false;
- private Map<String, String> _taskNameMap = Collections.emptyMap();
-
- public TaskConfig build() {
- validate();
-
- return new TaskConfig(_workflow, _targetResource, _targetPartitions, _targetPartitionStates,
- _command, _commandConfig, _timeoutPerPartition, _numConcurrentTasksPerInstance,
- _maxAttemptsPerPartition, _longLived, _taskNameMap);
- }
-
- /**
- * Convenience method to build a {@link TaskConfig} from a {@code Map<String, String>}.
- * @param cfg A map of property names to their string representations.
- * @return A {@link Builder}.
- */
- public static Builder fromMap(Map<String, String> cfg) {
- Builder b = new Builder();
- if (cfg.containsKey(WORKFLOW_ID)) {
- b.setWorkflow(cfg.get(WORKFLOW_ID));
- }
- if (cfg.containsKey(TARGET_RESOURCE)) {
- b.setTargetResource(cfg.get(TARGET_RESOURCE));
- }
- if (cfg.containsKey(TARGET_PARTITIONS)) {
- b.setTargetPartitions(csvToIntList(cfg.get(TARGET_PARTITIONS)));
- }
- if (cfg.containsKey(TARGET_PARTITION_STATES)) {
- b.setTargetPartitionStates(new HashSet<String>(Arrays.asList(cfg.get(
- TARGET_PARTITION_STATES).split(","))));
- }
- if (cfg.containsKey(COMMAND)) {
- b.setCommand(cfg.get(COMMAND));
- }
- if (cfg.containsKey(COMMAND_CONFIG)) {
- b.setCommandConfig(cfg.get(COMMAND_CONFIG));
- }
- if (cfg.containsKey(TIMEOUT_PER_PARTITION)) {
- b.setTimeoutPerPartition(Long.parseLong(cfg.get(TIMEOUT_PER_PARTITION)));
- }
- if (cfg.containsKey(NUM_CONCURRENT_TASKS_PER_INSTANCE)) {
- b.setNumConcurrentTasksPerInstance(Integer.parseInt(cfg
- .get(NUM_CONCURRENT_TASKS_PER_INSTANCE)));
- }
- if (cfg.containsKey(MAX_ATTEMPTS_PER_PARTITION)) {
- b.setMaxAttemptsPerPartition(Integer.parseInt(cfg.get(MAX_ATTEMPTS_PER_PARTITION)));
- }
- if (cfg.containsKey(LONG_LIVED)) {
- b.setLongLived(Boolean.parseBoolean(cfg.get(LONG_LIVED)));
- }
- return b;
- }
-
- public Builder setWorkflow(String v) {
- _workflow = v;
- return this;
- }
-
- public Builder setTargetResource(String v) {
- _targetResource = v;
- return this;
- }
-
- public Builder setTargetPartitions(List<Integer> v) {
- _targetPartitions = ImmutableList.copyOf(v);
- return this;
- }
-
- public Builder setTargetPartitionStates(Set<String> v) {
- _targetPartitionStates = ImmutableSet.copyOf(v);
- return this;
- }
-
- public Builder setCommand(String v) {
- _command = v;
- return this;
- }
-
- public Builder setCommandConfig(String v) {
- _commandConfig = v;
- return this;
- }
-
- public Builder setTimeoutPerPartition(long v) {
- _timeoutPerPartition = v;
- return this;
- }
-
- public Builder setNumConcurrentTasksPerInstance(int v) {
- _numConcurrentTasksPerInstance = v;
- return this;
- }
-
- public Builder setMaxAttemptsPerPartition(int v) {
- _maxAttemptsPerPartition = v;
- return this;
- }
-
- public Builder setLongLived(boolean isLongLived) {
- _longLived = isLongLived;
- return this;
- }
-
- public Builder setTaskNameMap(Map<String, String> taskNameMap) {
- _taskNameMap = taskNameMap;
- return this;
- }
-
- private void validate() {
- if (_targetResource == null && _targetPartitions == null) {
- throw new IllegalArgumentException(String.format(
- "%s cannot be null without specified partitions", TARGET_RESOURCE));
- }
- if (_targetResource != null && _targetPartitionStates != null
- && _targetPartitionStates.isEmpty()) {
- throw new IllegalArgumentException(String.format("%s cannot be empty",
- TARGET_PARTITION_STATES));
- }
- if (_command == null) {
- throw new IllegalArgumentException(String.format("%s cannot be null", COMMAND));
- }
- if (_timeoutPerPartition < 0) {
- throw new IllegalArgumentException(String.format("%s has invalid value %s",
- TIMEOUT_PER_PARTITION, _timeoutPerPartition));
- }
- if (_numConcurrentTasksPerInstance < 1) {
- throw new IllegalArgumentException(String.format("%s has invalid value %s",
- NUM_CONCURRENT_TASKS_PER_INSTANCE, _numConcurrentTasksPerInstance));
- }
- if (_maxAttemptsPerPartition < 1) {
- throw new IllegalArgumentException(String.format("%s has invalid value %s",
- MAX_ATTEMPTS_PER_PARTITION, _maxAttemptsPerPartition));
- }
- if (_workflow == null) {
- throw new IllegalArgumentException(String.format("%s cannot be null", WORKFLOW_ID));
- }
- }
-
- private static List<Integer> csvToIntList(String csv) {
- String[] vals = csv.split(",");
- List<Integer> l = new ArrayList<Integer>();
- for (String v : vals) {
- if (v != null && !v.isEmpty()) {
- l.add(Integer.parseInt(v));
- }
- }
+ public static TaskConfig from(TaskBean bean) {
+ return new TaskConfig(bean.command, bean.taskConfigMap);
+ }
- return l;
- }
+ /**
+ * Instantiate a typed configuration from a raw string map
+ * @param rawConfigMap mixed map of configuration and task metadata
+ * @return instantiated TaskConfig
+ */
+ public static TaskConfig from(Map<String, String> rawConfigMap) {
+ String taskId = rawConfigMap.get(TaskConfigFields.TASK_ID.toString());
+ String command = rawConfigMap.get(TaskConfigFields.TASK_COMMAND.toString());
+ return new TaskConfig(command, rawConfigMap, taskId);
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskContext.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskContext.java b/helix-core/src/main/java/org/apache/helix/task/TaskContext.java
deleted file mode 100644
index d416a86..0000000
--- a/helix-core/src/main/java/org/apache/helix/task/TaskContext.java
+++ /dev/null
@@ -1,135 +0,0 @@
-package org.apache.helix.task;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import java.util.Map;
-import java.util.TreeMap;
-import org.apache.helix.HelixProperty;
-import org.apache.helix.ZNRecord;
-
-/**
- * Provides a typed interface to the context information stored by {@link TaskRebalancer} in the
- * Helix property store.
- */
-public class TaskContext extends HelixProperty {
- public static final String START_TIME = "START_TIME";
- public static final String PARTITION_STATE = "STATE";
- public static final String NUM_ATTEMPTS = "NUM_ATTEMPTS";
- public static final String FINISH_TIME = "FINISH_TIME";
-
- public TaskContext(ZNRecord record) {
- super(record);
- }
-
- public void setStartTime(long t) {
- _record.setSimpleField(START_TIME, String.valueOf(t));
- }
-
- public long getStartTime() {
- String tStr = _record.getSimpleField(START_TIME);
- if (tStr == null) {
- return -1;
- }
-
- return Long.parseLong(tStr);
- }
-
- public void setPartitionState(int p, TaskPartitionState s) {
- String pStr = String.valueOf(p);
- Map<String, String> map = _record.getMapField(pStr);
- if (map == null) {
- map = new TreeMap<String, String>();
- _record.setMapField(pStr, map);
- }
- map.put(PARTITION_STATE, s.name());
- }
-
- public TaskPartitionState getPartitionState(int p) {
- Map<String, String> map = _record.getMapField(String.valueOf(p));
- if (map == null) {
- return null;
- }
-
- String str = map.get(PARTITION_STATE);
- if (str != null) {
- return TaskPartitionState.valueOf(str);
- } else {
- return null;
- }
- }
-
- public void setPartitionNumAttempts(int p, int n) {
- String pStr = String.valueOf(p);
- Map<String, String> map = _record.getMapField(pStr);
- if (map == null) {
- map = new TreeMap<String, String>();
- _record.setMapField(pStr, map);
- }
- map.put(NUM_ATTEMPTS, String.valueOf(n));
- }
-
- public int incrementNumAttempts(int pId) {
- int n = this.getPartitionNumAttempts(pId);
- if (n < 0) {
- n = 0;
- }
- n += 1;
- this.setPartitionNumAttempts(pId, n);
- return n;
- }
-
- public int getPartitionNumAttempts(int p) {
- Map<String, String> map = _record.getMapField(String.valueOf(p));
- if (map == null) {
- return -1;
- }
-
- String nStr = map.get(NUM_ATTEMPTS);
- if (nStr == null) {
- return -1;
- }
-
- return Integer.parseInt(nStr);
- }
-
- public void setPartitionFinishTime(int p, long t) {
- String pStr = String.valueOf(p);
- Map<String, String> map = _record.getMapField(pStr);
- if (map == null) {
- map = new TreeMap<String, String>();
- _record.setMapField(pStr, map);
- }
- map.put(FINISH_TIME, String.valueOf(t));
- }
-
- public long getPartitionFinishTime(int p) {
- Map<String, String> map = _record.getMapField(String.valueOf(p));
- if (map == null) {
- return -1;
- }
-
- String tStr = map.get(FINISH_TIME);
- if (tStr == null) {
- return -1;
- }
-
- return Long.parseLong(tStr);
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskDag.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskDag.java b/helix-core/src/main/java/org/apache/helix/task/TaskDag.java
deleted file mode 100644
index ab5bc62..0000000
--- a/helix-core/src/main/java/org/apache/helix/task/TaskDag.java
+++ /dev/null
@@ -1,152 +0,0 @@
-package org.apache.helix.task;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import org.codehaus.jackson.annotate.JsonProperty;
-import org.codehaus.jackson.map.ObjectMapper;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.TreeSet;
-
-/**
- * Provides a convenient way to construct, traverse,
- * and validate a task dependency graph
- */
-public class TaskDag {
- @JsonProperty("parentsToChildren")
- private final Map<String, Set<String>> _parentsToChildren;
-
- @JsonProperty("childrenToParents")
- private final Map<String, Set<String>> _childrenToParents;
-
- @JsonProperty("allNodes")
- private final Set<String> _allNodes;
-
- public static final TaskDag EMPTY_DAG = new TaskDag();
-
- public TaskDag() {
- _parentsToChildren = new TreeMap<String, Set<String>>();
- _childrenToParents = new TreeMap<String, Set<String>>();
- _allNodes = new TreeSet<String>();
- }
-
- public void addParentToChild(String parent, String child) {
- if (!_parentsToChildren.containsKey(parent)) {
- _parentsToChildren.put(parent, new TreeSet<String>());
- }
- _parentsToChildren.get(parent).add(child);
-
- if (!_childrenToParents.containsKey(child)) {
- _childrenToParents.put(child, new TreeSet<String>());
- }
- _childrenToParents.get(child).add(parent);
-
- _allNodes.add(parent);
- _allNodes.add(child);
- }
-
- public void addNode(String node) {
- _allNodes.add(node);
- }
-
- public Map<String, Set<String>> getParentsToChildren() {
- return _parentsToChildren;
- }
-
- public Map<String, Set<String>> getChildrenToParents() {
- return _childrenToParents;
- }
-
- public Set<String> getAllNodes() {
- return _allNodes;
- }
-
- public Set<String> getDirectChildren(String node) {
- if (!_parentsToChildren.containsKey(node)) {
- return Collections.emptySet();
- }
- return _parentsToChildren.get(node);
- }
-
- public Set<String> getDirectParents(String node) {
- if (!_childrenToParents.containsKey(node)) {
- return Collections.emptySet();
- }
- return _childrenToParents.get(node);
- }
-
- public String toJson() throws Exception {
- return new ObjectMapper().writeValueAsString(this);
- }
-
- public static TaskDag fromJson(String json) {
- try {
- return new ObjectMapper().readValue(json, TaskDag.class);
- } catch (Exception e) {
- throw new IllegalArgumentException("Unable to parse json " + json + " into task dag");
- }
- }
-
- /**
- * Checks that dag contains no cycles and all nodes are reachable.
- */
- public void validate() {
- Set<String> prevIteration = new TreeSet<String>();
-
- // get all unparented nodes
- for (String node : _allNodes) {
- if (getDirectParents(node).isEmpty()) {
- prevIteration.add(node);
- }
- }
-
- // visit children nodes up to max iteration count, by which point we should have exited
- // naturally
- Set<String> allNodesReached = new TreeSet<String>();
- int iterationCount = 0;
- int maxIterations = _allNodes.size() + 1;
-
- while (!prevIteration.isEmpty() && iterationCount < maxIterations) {
- // construct set of all children reachable from prev iteration
- Set<String> thisIteration = new TreeSet<String>();
- for (String node : prevIteration) {
- thisIteration.addAll(getDirectChildren(node));
- }
-
- allNodesReached.addAll(prevIteration);
- prevIteration = thisIteration;
- iterationCount++;
- }
-
- allNodesReached.addAll(prevIteration);
-
- if (iterationCount >= maxIterations) {
- throw new IllegalArgumentException("DAG invalid: cycles detected");
- }
-
- if (!allNodesReached.containsAll(_allNodes)) {
- throw new IllegalArgumentException("DAG invalid: unreachable nodes found. Reachable set is "
- + allNodesReached);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/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 dd47625..ada2f99 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
@@ -20,8 +20,8 @@ package org.apache.helix.task;
*/
import java.io.File;
-import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@@ -41,11 +41,13 @@ import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerFactory;
import org.apache.helix.HelixProperty;
import org.apache.helix.InstanceType;
-import org.apache.helix.controller.rebalancer.HelixRebalancer;
+import org.apache.helix.PropertyKey;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.builder.CustomModeISBuilder;
import org.apache.log4j.Logger;
+import com.beust.jcommander.internal.Lists;
+
/**
* CLI for scheduling/canceling workflows
*/
@@ -152,67 +154,77 @@ public class TaskDriver {
flow.getResourceConfigMap());
// then schedule tasks
- for (String task : flow.getTaskConfigs().keySet()) {
- scheduleTask(task, TaskConfig.Builder.fromMap(flow.getTaskConfigs().get(task)).build());
+ for (String job : flow.getJobConfigs().keySet()) {
+ JobConfig.Builder builder = JobConfig.Builder.fromMap(flow.getJobConfigs().get(job));
+ if (flow.getTaskConfigs() != null && flow.getTaskConfigs().containsKey(job)) {
+ builder.addTaskConfigs(flow.getTaskConfigs().get(job));
+ }
+ scheduleJob(job, builder.build());
}
}
- /** Posts new task to cluster */
- private void scheduleTask(String taskResource, TaskConfig taskConfig) throws Exception {
- // Set up task resource based on partitions provided, or from target resource
- int numPartitions;
- List<Integer> partitions = taskConfig.getTargetPartitions();
- String targetResource = taskConfig.getTargetResource();
- if (partitions != null && !partitions.isEmpty()) {
- numPartitions = partitions.size();
- } else if (targetResource != null) {
- numPartitions =
- _admin.getResourceIdealState(_clusterName, taskConfig.getTargetResource())
- .getPartitionSet().size();
- } else {
- numPartitions = 0;
+ /** Posts new job to cluster */
+ private void scheduleJob(String jobResource, JobConfig jobConfig) throws Exception {
+ // Set up job resource based on partitions from target resource
+ int numIndependentTasks = jobConfig.getTaskConfigMap().size();
+ int numPartitions =
+ (numIndependentTasks > 0) ? numIndependentTasks : _admin
+ .getResourceIdealState(_clusterName, jobConfig.getTargetResource()).getPartitionSet()
+ .size();
+ _admin.addResource(_clusterName, jobResource, numPartitions, TaskConstants.STATE_MODEL_NAME);
+
+ // Set the job configuration
+ HelixDataAccessor accessor = _manager.getHelixDataAccessor();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ HelixProperty resourceConfig = new HelixProperty(jobResource);
+ resourceConfig.getRecord().getSimpleFields().putAll(jobConfig.getResourceConfigMap());
+ Map<String, TaskConfig> taskConfigMap = jobConfig.getTaskConfigMap();
+ if (taskConfigMap != null) {
+ for (TaskConfig taskConfig : taskConfigMap.values()) {
+ resourceConfig.getRecord().setMapField(taskConfig.getId(), taskConfig.getConfigMap());
+ }
}
- _admin.addResource(_clusterName, taskResource, numPartitions, TaskConstants.STATE_MODEL_NAME);
- _admin.setConfig(TaskUtil.getResourceConfigScope(_clusterName, taskResource),
- taskConfig.getResourceConfigMap());
+ accessor.setProperty(keyBuilder.resourceConfig(jobResource), resourceConfig);
// Push out new ideal state based on number of target partitions
- CustomModeISBuilder builder = new CustomModeISBuilder(taskResource);
+ CustomModeISBuilder builder = new CustomModeISBuilder(jobResource);
builder.setRebalancerMode(IdealState.RebalanceMode.USER_DEFINED);
builder.setNumReplica(1);
builder.setNumPartitions(numPartitions);
builder.setStateModel(TaskConstants.STATE_MODEL_NAME);
for (int i = 0; i < numPartitions; i++) {
- builder.add(taskResource + "_" + i);
+ builder.add(jobResource + "_" + i);
}
IdealState is = builder.build();
- Class<? extends HelixRebalancer> rebalancerClass =
- (targetResource != null) ? TaskRebalancer.class : IndependentTaskRebalancer.class;
- is.setRebalancerClassName(rebalancerClass.getName());
- _admin.setResourceIdealState(_clusterName, taskResource, is);
+ if (taskConfigMap != null && !taskConfigMap.isEmpty()) {
+ is.setRebalancerClassName(GenericTaskRebalancer.class.getName());
+ } else {
+ is.setRebalancerClassName(FixedTargetTaskRebalancer.class.getName());
+ }
+ _admin.setResourceIdealState(_clusterName, jobResource, is);
}
- /** Public method to resume a task/workflow */
+ /** Public method to resume a job/workflow */
public void resume(String resource) {
setTaskTargetState(resource, TargetState.START);
}
- /** Public method to stop a task/workflow */
+ /** Public method to stop a job/workflow */
public void stop(String resource) {
setTaskTargetState(resource, TargetState.STOP);
}
- /** Public method to delete a task/workflow */
+ /** Public method to delete a job/workflow */
public void delete(String resource) {
setTaskTargetState(resource, TargetState.DELETE);
}
/** Helper function to change target state for a given task */
- private void setTaskTargetState(String taskResource, TargetState state) {
+ private void setTaskTargetState(String jobResource, TargetState state) {
HelixDataAccessor accessor = _manager.getHelixDataAccessor();
- HelixProperty p = new HelixProperty(taskResource);
+ HelixProperty p = new HelixProperty(jobResource);
p.getRecord().setSimpleField(WorkflowConfig.TARGET_STATE, state.name());
- accessor.updateProperty(accessor.keyBuilder().resourceConfig(taskResource), p);
+ accessor.updateProperty(accessor.keyBuilder().resourceConfig(jobResource), p);
invokeRebalance();
}
@@ -222,34 +234,24 @@ public class TaskDriver {
WorkflowContext wCtx = TaskUtil.getWorkflowContext(_manager, resource);
LOG.info("Workflow " + resource + " consists of the following tasks: "
- + wCfg.getTaskDag().getAllNodes());
+ + wCfg.getJobDag().getAllNodes());
LOG.info("Current state of workflow is " + wCtx.getWorkflowState().name());
- LOG.info("Task states are: ");
+ LOG.info("Job states are: ");
LOG.info("-------");
- for (String task : wCfg.getTaskDag().getAllNodes()) {
- LOG.info("Task " + task + " is " + wCtx.getTaskState(task));
+ for (String job : wCfg.getJobDag().getAllNodes()) {
+ LOG.info("Task " + job + " is " + wCtx.getJobState(job));
// fetch task information
- TaskContext tCtx = TaskUtil.getTaskContext(_manager, task);
- TaskConfig tCfg = TaskUtil.getTaskCfg(_manager, task);
+ JobContext jCtx = TaskUtil.getJobContext(_manager, job);
// calculate taskPartitions
- List<Integer> partitions;
- if (tCfg.getTargetPartitions() != null) {
- partitions = tCfg.getTargetPartitions();
- } else {
- partitions = new ArrayList<Integer>();
- for (String pStr : _admin.getResourceIdealState(_clusterName, tCfg.getTargetResource())
- .getPartitionSet()) {
- partitions
- .add(Integer.parseInt(pStr.substring(pStr.lastIndexOf("_") + 1, pStr.length())));
- }
- }
+ List<Integer> partitions = Lists.newArrayList(jCtx.getPartitionSet());
+ Collections.sort(partitions);
// group partitions by status
Map<TaskPartitionState, Integer> statusCount = new TreeMap<TaskPartitionState, Integer>();
for (Integer i : partitions) {
- TaskPartitionState s = tCtx.getPartitionState(i);
+ TaskPartitionState s = jCtx.getPartitionState(i);
if (!statusCount.containsKey(s)) {
statusCount.put(s, 0);
}
@@ -288,20 +290,26 @@ public class TaskDriver {
return options;
}
- /** Constructs option group containing options required by all drivable tasks */
+ /** Constructs option group containing options required by all drivable jobs */
@SuppressWarnings("static-access")
private static OptionGroup contructGenericRequiredOptionGroup() {
Option zkAddressOption =
- OptionBuilder.isRequired().hasArgs(1).withArgName("zkAddress").withLongOpt(ZK_ADDRESS)
- .withDescription("ZK address managing target cluster").create();
+ OptionBuilder.isRequired().withLongOpt(ZK_ADDRESS)
+ .withDescription("ZK address managing cluster").create();
+ zkAddressOption.setArgs(1);
+ zkAddressOption.setArgName("zkAddress");
Option clusterNameOption =
- OptionBuilder.isRequired().hasArgs(1).withArgName("clusterName")
- .withLongOpt(CLUSTER_NAME_OPTION).withDescription("Target cluster name").create();
+ OptionBuilder.isRequired().withLongOpt(CLUSTER_NAME_OPTION).withDescription("Cluster name")
+ .create();
+ clusterNameOption.setArgs(1);
+ clusterNameOption.setArgName("clusterName");
Option taskResourceOption =
- OptionBuilder.isRequired().hasArgs(1).withArgName("resourceName")
- .withLongOpt(RESOURCE_OPTION).withDescription("Target workflow or task").create();
+ OptionBuilder.isRequired().withLongOpt(RESOURCE_OPTION)
+ .withDescription("Workflow or job name").create();
+ taskResourceOption.setArgs(1);
+ taskResourceOption.setArgName("resourceName");
OptionGroup group = new OptionGroup();
group.addOption(zkAddressOption);
@@ -310,12 +318,14 @@ public class TaskDriver {
return group;
}
- /** Constructs option group containing options required by all drivable tasks */
- @SuppressWarnings("static-access")
+ /** Constructs option group containing options required by all drivable jobs */
private static OptionGroup constructStartOptionGroup() {
+ @SuppressWarnings("static-access")
Option workflowFileOption =
- OptionBuilder.withLongOpt(WORKFLOW_FILE_OPTION).hasArgs(1).withArgName("workflowFile")
+ OptionBuilder.withLongOpt(WORKFLOW_FILE_OPTION)
.withDescription("Local file describing workflow").create();
+ workflowFileOption.setArgs(1);
+ workflowFileOption.setArgName("workflowFile");
OptionGroup group = new OptionGroup();
group.addOption(workflowFileOption);
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskFactory.java b/helix-core/src/main/java/org/apache/helix/task/TaskFactory.java
index 0cbf24c..31fddc7 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskFactory.java
@@ -19,14 +19,15 @@ package org.apache.helix.task;
* under the License.
*/
+
/**
* A factory for {@link Task} objects.
*/
public interface TaskFactory {
/**
* Returns a {@link Task} instance.
- * @param config Configuration information for the task.
+ * @param context Contextual information for the task, including task and job configurations
* @return A {@link Task} instance.
*/
- Task createNewTask(String config);
+ Task createNewTask(TaskCallbackContext context);
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskRebalancer.java b/helix-core/src/main/java/org/apache/helix/task/TaskRebalancer.java
index 7b93b82..829f0c4 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskRebalancer.java
@@ -19,112 +19,585 @@ package org.apache.helix.task;
* under the License.
*/
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
+import java.util.TreeMap;
import java.util.TreeSet;
+import org.apache.helix.AccessOption;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.ZNRecord;
import org.apache.helix.api.Cluster;
import org.apache.helix.api.Resource;
import org.apache.helix.api.State;
import org.apache.helix.api.id.ParticipantId;
import org.apache.helix.api.id.PartitionId;
import org.apache.helix.api.id.ResourceId;
+import org.apache.helix.controller.context.ControllerContextProvider;
+import org.apache.helix.controller.rebalancer.HelixRebalancer;
+import org.apache.helix.controller.rebalancer.config.RebalancerConfig;
import org.apache.helix.controller.stages.ResourceCurrentState;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.ResourceAssignment;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
/**
- * Custom rebalancer implementation for the {@code Task} state model. Tasks are assigned to
- * instances hosting target resource partitions in target states
+ * Custom rebalancer implementation for the {@code Task} state model.
*/
-public class TaskRebalancer extends AbstractTaskRebalancer {
+public abstract class TaskRebalancer implements HelixRebalancer {
+ private static final Logger LOG = Logger.getLogger(TaskRebalancer.class);
+ private HelixManager _manager;
+
+ /**
+ * Get all the partitions that should be created by this task
+ * @param jobCfg the task configuration
+ * @param jobCtx the task context
+ * @param workflowCfg the workflow configuration
+ * @param workflowCtx the workflow context
+ * @param cache cluster snapshot
+ * @return set of partition numbers
+ */
+ public abstract Set<Integer> getAllTaskPartitions(JobConfig jobCfg, JobContext jobCtx,
+ WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Cluster cache);
+
+ /**
+ * Compute an assignment of tasks to instances
+ * @param currStateOutput the current state of the instances
+ * @param prevAssignment the previous task partition assignment
+ * @param instanceList the instances
+ * @param jobCfg the task configuration
+ * @param taskCtx the task context
+ * @param workflowCfg the workflow configuration
+ * @param workflowCtx the workflow context
+ * @param partitionSet the partitions to assign
+ * @param cache cluster snapshot
+ * @return map of instances to set of partition numbers
+ */
+ public abstract Map<ParticipantId, SortedSet<Integer>> getTaskAssignment(
+ ResourceCurrentState currStateOutput, ResourceAssignment prevAssignment,
+ Iterable<ParticipantId> instanceList, JobConfig jobCfg, JobContext jobContext,
+ WorkflowConfig workflowCfg, WorkflowContext workflowCtx, Set<Integer> partitionSet,
+ Cluster cache);
+
@Override
- public Set<Integer> getAllTaskPartitions(TaskConfig taskCfg, WorkflowConfig workflowCfg,
- WorkflowContext workflowCtx, Cluster cluster) {
- return getAllTaskPartitions(getTgtIdealState(taskCfg, cluster), taskCfg);
+ public void init(HelixManager manager, ControllerContextProvider contextProvider) {
+ _manager = manager;
}
@Override
- public Map<String, SortedSet<Integer>> getTaskAssignment(ResourceCurrentState currStateOutput,
- ResourceAssignment prevAssignment, Iterable<ParticipantId> instanceList, TaskConfig taskCfg,
- TaskContext taskCtx, WorkflowConfig workflowCfg, WorkflowContext workflowCtx,
- Set<Integer> partitionSet, Cluster cluster) {
- IdealState tgtIs = getTgtIdealState(taskCfg, cluster);
- if (tgtIs == null) {
- return Collections.emptyMap();
+ public ResourceAssignment computeResourceMapping(RebalancerConfig rebalancerConfig,
+ ResourceAssignment prevAssignment, Cluster cluster, ResourceCurrentState currentState) {
+ IdealState taskIs = cluster.getResource(rebalancerConfig.getResourceId()).getIdealState();
+ return computeBestPossiblePartitionState(cluster, taskIs,
+ cluster.getResource(rebalancerConfig.getResourceId()), currentState);
+ }
+
+ public ResourceAssignment computeBestPossiblePartitionState(Cluster clusterData,
+ IdealState taskIs, Resource resource, ResourceCurrentState currStateOutput) {
+ final String resourceName = resource.getId().toString();
+
+ // Fetch job configuration
+ JobConfig jobCfg = TaskUtil.getJobCfg(_manager, resourceName);
+ String workflowResource = jobCfg.getWorkflow();
+
+ // Fetch workflow configuration and context
+ WorkflowConfig workflowCfg = TaskUtil.getWorkflowCfg(_manager, workflowResource);
+ WorkflowContext workflowCtx = TaskUtil.getWorkflowContext(_manager, workflowResource);
+
+ // Initialize workflow context if needed
+ if (workflowCtx == null) {
+ workflowCtx = new WorkflowContext(new ZNRecord("WorkflowContext"));
+ workflowCtx.setStartTime(System.currentTimeMillis());
+ }
+
+ // Check parent dependencies
+ for (String parent : workflowCfg.getJobDag().getDirectParents(resourceName)) {
+ if (workflowCtx.getJobState(parent) == null
+ || !workflowCtx.getJobState(parent).equals(TaskState.COMPLETED)) {
+ return emptyAssignment(resourceName);
+ }
+ }
+
+ // Clean up if workflow marked for deletion
+ TargetState targetState = workflowCfg.getTargetState();
+ if (targetState == TargetState.DELETE) {
+ cleanup(_manager, resourceName, workflowCfg, workflowResource);
+ return emptyAssignment(resourceName);
+ }
+
+ // Check if this workflow has been finished past its expiry.
+ if (workflowCtx.getFinishTime() != WorkflowContext.UNFINISHED
+ && workflowCtx.getFinishTime() + workflowCfg.getExpiry() <= System.currentTimeMillis()) {
+ markForDeletion(_manager, workflowResource);
+ cleanup(_manager, resourceName, workflowCfg, workflowResource);
+ return emptyAssignment(resourceName);
+ }
+
+ // Fetch any existing context information from the property store.
+ JobContext jobCtx = TaskUtil.getJobContext(_manager, resourceName);
+ if (jobCtx == null) {
+ jobCtx = new JobContext(new ZNRecord("TaskContext"));
+ jobCtx.setStartTime(System.currentTimeMillis());
+ }
+
+ // The job is already in a final state (completed/failed).
+ if (workflowCtx.getJobState(resourceName) == TaskState.FAILED
+ || workflowCtx.getJobState(resourceName) == TaskState.COMPLETED) {
+ return emptyAssignment(resourceName);
+ }
+
+ ResourceAssignment prevAssignment = TaskUtil.getPrevResourceAssignment(_manager, resourceName);
+ if (prevAssignment == null) {
+ prevAssignment = new ResourceAssignment(ResourceId.from(resourceName));
+ }
+
+ // Will contain the list of partitions that must be explicitly dropped from the ideal state that
+ // is stored in zk.
+ // Fetch the previous resource assignment from the property store. This is required because of
+ // HELIX-230.
+ Set<Integer> partitionsToDrop = new TreeSet<Integer>();
+
+ ResourceAssignment newAssignment =
+ computeResourceMapping(resourceName, workflowCfg, jobCfg, prevAssignment, clusterData
+ .getLiveParticipantMap().keySet(), currStateOutput, workflowCtx, jobCtx,
+ partitionsToDrop, clusterData);
+
+ if (!partitionsToDrop.isEmpty()) {
+ for (Integer pId : partitionsToDrop) {
+ taskIs.getRecord().getMapFields().remove(pName(resourceName, pId));
+ }
+ HelixDataAccessor accessor = _manager.getHelixDataAccessor();
+ PropertyKey propertyKey = accessor.keyBuilder().idealStates(resourceName);
+ accessor.setProperty(propertyKey, taskIs);
+ }
+
+ // Update rebalancer context, previous ideal state.
+ TaskUtil.setJobContext(_manager, resourceName, jobCtx);
+ TaskUtil.setWorkflowContext(_manager, workflowResource, workflowCtx);
+ TaskUtil.setPrevResourceAssignment(_manager, resourceName, newAssignment);
+
+ return newAssignment;
+ }
+
+ private ResourceAssignment computeResourceMapping(String jobResource,
+ WorkflowConfig workflowConfig, JobConfig jobCfg, ResourceAssignment prevAssignment,
+ Iterable<ParticipantId> liveInstances, ResourceCurrentState currStateOutput,
+ WorkflowContext workflowCtx, JobContext jobCtx, Set<Integer> partitionsToDropFromIs,
+ Cluster cache) {
+ TargetState jobTgtState = workflowConfig.getTargetState();
+
+ // Update running status in workflow context
+ if (jobTgtState == TargetState.STOP) {
+ workflowCtx.setJobState(jobResource, TaskState.STOPPED);
+ // Workflow has been stopped if all jobs are stopped
+ if (isWorkflowStopped(workflowCtx, workflowConfig)) {
+ workflowCtx.setWorkflowState(TaskState.STOPPED);
+ }
+ } else {
+ workflowCtx.setJobState(jobResource, TaskState.IN_PROGRESS);
+ // Workflow is in progress if any task is in progress
+ workflowCtx.setWorkflowState(TaskState.IN_PROGRESS);
+ }
+
+ // Used to keep track of tasks that have already been assigned to instances.
+ Set<Integer> assignedPartitions = new HashSet<Integer>();
+
+ // Keeps a mapping of (partition) -> (instance, state)
+ Map<Integer, PartitionAssignment> paMap = new TreeMap<Integer, PartitionAssignment>();
+
+ // Process all the current assignments of tasks.
+ Set<Integer> allPartitions =
+ getAllTaskPartitions(jobCfg, jobCtx, workflowConfig, workflowCtx, cache);
+ Map<ParticipantId, SortedSet<Integer>> taskAssignments =
+ getTaskPartitionAssignments(liveInstances, prevAssignment, allPartitions);
+ for (ParticipantId instance : taskAssignments.keySet()) {
+ Set<Integer> pSet = taskAssignments.get(instance);
+ // Used to keep track of partitions that are in one of the final states: COMPLETED, TIMED_OUT,
+ // TASK_ERROR, ERROR.
+ Set<Integer> donePartitions = new TreeSet<Integer>();
+ for (int pId : pSet) {
+ final String pName = pName(jobResource, pId);
+
+ // Check for pending state transitions on this (partition, instance).
+ State pendingState =
+ currStateOutput.getPendingState(ResourceId.from(jobResource), PartitionId.from(pName),
+ instance);
+ if (pendingState != null) {
+ // There is a pending state transition for this (partition, instance). Just copy forward
+ // the state
+ // assignment from the previous ideal state.
+ Map<ParticipantId, State> stateMap =
+ prevAssignment.getReplicaMap(PartitionId.from(pName));
+ if (stateMap != null) {
+ State prevState = stateMap.get(instance);
+ paMap.put(pId, new PartitionAssignment(instance.toString(), prevState.toString()));
+ assignedPartitions.add(pId);
+ LOG.debug(String
+ .format(
+ "Task partition %s has a pending state transition on instance %s. Using the previous ideal state which was %s.",
+ pName, instance, prevState));
+ }
+
+ continue;
+ }
+
+ State currHelixState =
+ currStateOutput.getCurrentState(ResourceId.from(jobResource), PartitionId.from(pName),
+ instance);
+ TaskPartitionState currState =
+ (currHelixState != null) ? TaskPartitionState.valueOf(currHelixState.toString()) : null;
+
+ // Process any requested state transitions.
+ State requestedStateStr =
+ currStateOutput.getRequestedState(ResourceId.from(jobResource),
+ PartitionId.from(pName), instance);
+ if (requestedStateStr != null && !requestedStateStr.toString().isEmpty()) {
+ TaskPartitionState requestedState =
+ TaskPartitionState.valueOf(requestedStateStr.toString());
+ if (requestedState.equals(currState)) {
+ LOG.warn(String.format(
+ "Requested state %s is the same as the current state for instance %s.",
+ requestedState, instance));
+ }
+
+ paMap.put(pId, new PartitionAssignment(instance.toString(), requestedState.name()));
+ assignedPartitions.add(pId);
+ LOG.debug(String.format(
+ "Instance %s requested a state transition to %s for partition %s.", instance,
+ requestedState, pName));
+ continue;
+ }
+
+ switch (currState) {
+ case RUNNING:
+ case STOPPED: {
+ TaskPartitionState nextState;
+ if (jobTgtState == TargetState.START) {
+ nextState = TaskPartitionState.RUNNING;
+ } else {
+ nextState = TaskPartitionState.STOPPED;
+ }
+
+ paMap.put(pId, new PartitionAssignment(instance.toString(), nextState.name()));
+ assignedPartitions.add(pId);
+ LOG.debug(String.format("Setting task partition %s state to %s on instance %s.", pName,
+ nextState, instance));
+ }
+ break;
+ case COMPLETED: {
+ // The task has completed on this partition. Mark as such in the context object.
+ donePartitions.add(pId);
+ LOG.debug(String
+ .format(
+ "Task partition %s has completed with state %s. Marking as such in rebalancer context.",
+ pName, currState));
+ partitionsToDropFromIs.add(pId);
+ markPartitionCompleted(jobCtx, pId);
+ }
+ break;
+ case TIMED_OUT:
+ case TASK_ERROR:
+ case ERROR: {
+ donePartitions.add(pId); // The task may be rescheduled on a different instance.
+ LOG.debug(String.format(
+ "Task partition %s has error state %s. Marking as such in rebalancer context.",
+ pName, currState));
+ markPartitionError(jobCtx, pId, currState);
+ // The error policy is to fail the task as soon a single partition fails for a specified
+ // maximum number of
+ // attempts.
+ if (jobCtx.getPartitionNumAttempts(pId) >= jobCfg.getMaxAttemptsPerTask()) {
+ workflowCtx.setJobState(jobResource, TaskState.FAILED);
+ workflowCtx.setWorkflowState(TaskState.FAILED);
+ addAllPartitions(allPartitions, partitionsToDropFromIs);
+ return emptyAssignment(jobResource);
+ }
+ }
+ break;
+ case INIT:
+ case DROPPED: {
+ // currState in [INIT, DROPPED]. Do nothing, the partition is eligible to be reassigned.
+ donePartitions.add(pId);
+ LOG.debug(String.format(
+ "Task partition %s has state %s. It will be dropped from the current ideal state.",
+ pName, currState));
+ }
+ break;
+ default:
+ throw new AssertionError("Unknown enum symbol: " + currState);
+ }
+ }
+
+ // Remove the set of task partitions that are completed or in one of the error states.
+ pSet.removeAll(donePartitions);
+ }
+
+ if (isJobComplete(jobCtx, allPartitions)) {
+ workflowCtx.setJobState(jobResource, TaskState.COMPLETED);
+ if (isWorkflowComplete(workflowCtx, workflowConfig)) {
+ workflowCtx.setWorkflowState(TaskState.COMPLETED);
+ workflowCtx.setFinishTime(System.currentTimeMillis());
+ }
+ }
+
+ // Make additional task assignments if needed.
+ if (jobTgtState == TargetState.START) {
+ // Contains the set of task partitions that must be excluded from consideration when making
+ // any new assignments.
+ // This includes all completed, failed, already assigned partitions.
+ Set<Integer> excludeSet = Sets.newTreeSet(assignedPartitions);
+ addCompletedPartitions(excludeSet, jobCtx, allPartitions);
+ // Get instance->[partition, ...] mappings for the target resource.
+ Map<ParticipantId, SortedSet<Integer>> tgtPartitionAssignments =
+ getTaskAssignment(currStateOutput, prevAssignment, liveInstances, jobCfg, jobCtx,
+ workflowConfig, workflowCtx, allPartitions, cache);
+ for (Map.Entry<ParticipantId, SortedSet<Integer>> entry : taskAssignments.entrySet()) {
+ ParticipantId instance = entry.getKey();
+ if (!tgtPartitionAssignments.containsKey(instance)) {
+ continue;
+ }
+ // Contains the set of task partitions currently assigned to the instance.
+ Set<Integer> pSet = entry.getValue();
+ int numToAssign = jobCfg.getNumConcurrentTasksPerInstance() - pSet.size();
+ if (numToAssign > 0) {
+ List<Integer> nextPartitions =
+ getNextPartitions(tgtPartitionAssignments.get(instance), excludeSet, numToAssign);
+ for (Integer pId : nextPartitions) {
+ String pName = pName(jobResource, pId);
+ paMap.put(pId,
+ new PartitionAssignment(instance.toString(), TaskPartitionState.RUNNING.name()));
+ excludeSet.add(pId);
+ LOG.debug(String.format("Setting task partition %s state to %s on instance %s.", pName,
+ TaskPartitionState.RUNNING, instance));
+ }
+ }
+ }
}
- Set<String> tgtStates = taskCfg.getTargetPartitionStates();
- return getTgtPartitionAssignment(currStateOutput, instanceList, tgtIs, tgtStates, partitionSet);
+
+ // Construct a ResourceAssignment object from the map of partition assignments.
+ ResourceAssignment ra = new ResourceAssignment(ResourceId.from(jobResource));
+ for (Map.Entry<Integer, PartitionAssignment> e : paMap.entrySet()) {
+ PartitionAssignment pa = e.getValue();
+ ra.addReplicaMap(PartitionId.from(pName(jobResource, e.getKey())),
+ ImmutableMap.of(ParticipantId.from(pa._instance), State.from(pa._state)));
+ }
+
+ return ra;
}
/**
- * Gets the ideal state of the target resource of this task
- * @param taskCfg task config containing target resource id
- * @param cluster snapshot of the cluster containing the task and target resource
- * @return target resource ideal state, or null
+ * Checks if the job has completed.
+ * @param ctx The rebalancer context.
+ * @param allPartitions The set of partitions to check.
+ * @return true if all task partitions have been marked with status
+ * {@link TaskPartitionState#COMPLETED} in the rebalancer
+ * context, false otherwise.
*/
- private static IdealState getTgtIdealState(TaskConfig taskCfg, Cluster cluster) {
- ResourceId tgtResourceId = ResourceId.from(taskCfg.getTargetResource());
- Resource resource = cluster.getResource(tgtResourceId);
- return resource != null ? resource.getIdealState() : null;
+ private static boolean isJobComplete(JobContext ctx, Set<Integer> allPartitions) {
+ for (Integer pId : allPartitions) {
+ TaskPartitionState state = ctx.getPartitionState(pId);
+ if (state != TaskPartitionState.COMPLETED) {
+ return false;
+ }
+ }
+ return true;
}
/**
- * Returns the set of all partition ids for a task.
- * <p/>
- * If a set of partition ids was explicitly specified in the config, that is used. Otherwise, we
- * use the list of all partition ids from the target resource.
+ * Checks if the workflow has completed.
+ * @param ctx Workflow context containing job states
+ * @param cfg Workflow config containing set of jobs
+ * @return returns true if all tasks are {@link TaskState#COMPLETED}, false otherwise.
*/
- private static Set<Integer> getAllTaskPartitions(IdealState tgtResourceIs, TaskConfig taskCfg) {
- if (tgtResourceIs == null) {
- return null;
+ private static boolean isWorkflowComplete(WorkflowContext ctx, WorkflowConfig cfg) {
+ for (String job : cfg.getJobDag().getAllNodes()) {
+ if (ctx.getJobState(job) != TaskState.COMPLETED) {
+ return false;
+ }
}
- Set<Integer> taskPartitions = new HashSet<Integer>();
- if (taskCfg.getTargetPartitions() != null) {
- for (Integer pId : taskCfg.getTargetPartitions()) {
- taskPartitions.add(pId);
+ return true;
+ }
+
+ /**
+ * Checks if the workflow has been stopped.
+ * @param ctx Workflow context containing task states
+ * @param cfg Workflow config containing set of tasks
+ * @return returns true if all tasks are {@link TaskState#STOPPED}, false otherwise.
+ */
+ private static boolean isWorkflowStopped(WorkflowContext ctx, WorkflowConfig cfg) {
+ for (String job : cfg.getJobDag().getAllNodes()) {
+ if (ctx.getJobState(job) != TaskState.STOPPED && ctx.getJobState(job) != null) {
+ return false;
}
- } else {
- for (String pName : tgtResourceIs.getPartitionSet()) {
- taskPartitions.add(pId(pName));
+ }
+ return true;
+ }
+
+ private static void markForDeletion(HelixManager mgr, String resourceName) {
+ mgr.getConfigAccessor().set(
+ TaskUtil.getResourceConfigScope(mgr.getClusterName(), resourceName),
+ WorkflowConfig.TARGET_STATE, TargetState.DELETE.name());
+ }
+
+ /**
+ * Cleans up all Helix state associated with this job, wiping workflow-level information if this
+ * is the last remaining job in its workflow.
+ */
+ private static void cleanup(HelixManager mgr, String resourceName, WorkflowConfig cfg,
+ String workflowResource) {
+ HelixDataAccessor accessor = mgr.getHelixDataAccessor();
+ // Delete resource configs.
+ PropertyKey cfgKey = getConfigPropertyKey(accessor, resourceName);
+ if (!accessor.removeProperty(cfgKey)) {
+ throw new RuntimeException(
+ String
+ .format(
+ "Error occurred while trying to clean up task %s. Failed to remove node %s from Helix. Aborting further clean up steps.",
+ resourceName, cfgKey));
+ }
+ // Delete property store information for this resource.
+ String propStoreKey = getRebalancerPropStoreKey(resourceName);
+ if (!mgr.getHelixPropertyStore().remove(propStoreKey, AccessOption.PERSISTENT)) {
+ throw new RuntimeException(
+ String
+ .format(
+ "Error occurred while trying to clean up task %s. Failed to remove node %s from Helix. Aborting further clean up steps.",
+ resourceName, propStoreKey));
+ }
+ // Finally, delete the ideal state itself.
+ PropertyKey isKey = getISPropertyKey(accessor, resourceName);
+ if (!accessor.removeProperty(isKey)) {
+ throw new RuntimeException(String.format(
+ "Error occurred while trying to clean up task %s. Failed to remove node %s from Helix.",
+ resourceName, isKey));
+ }
+ LOG.info(String.format("Successfully cleaned up task resource %s.", resourceName));
+
+ boolean lastInWorkflow = true;
+ for (String job : cfg.getJobDag().getAllNodes()) {
+ // check if property store information or resource configs exist for this job
+ if (mgr.getHelixPropertyStore().exists(getRebalancerPropStoreKey(job),
+ AccessOption.PERSISTENT)
+ || accessor.getProperty(getConfigPropertyKey(accessor, job)) != null
+ || accessor.getProperty(getISPropertyKey(accessor, job)) != null) {
+ lastInWorkflow = false;
+ }
+ }
+
+ // clean up job-level info if this was the last in workflow
+ if (lastInWorkflow) {
+ // delete workflow config
+ PropertyKey workflowCfgKey = getConfigPropertyKey(accessor, workflowResource);
+ if (!accessor.removeProperty(workflowCfgKey)) {
+ throw new RuntimeException(
+ String
+ .format(
+ "Error occurred while trying to clean up workflow %s. Failed to remove node %s from Helix. Aborting further clean up steps.",
+ workflowResource, workflowCfgKey));
+ }
+ // Delete property store information for this workflow
+ String workflowPropStoreKey = getRebalancerPropStoreKey(workflowResource);
+ if (!mgr.getHelixPropertyStore().remove(workflowPropStoreKey, AccessOption.PERSISTENT)) {
+ throw new RuntimeException(
+ String
+ .format(
+ "Error occurred while trying to clean up workflow %s. Failed to remove node %s from Helix. Aborting further clean up steps.",
+ workflowResource, workflowPropStoreKey));
+ }
+ }
+
+ }
+
+ private static String getRebalancerPropStoreKey(String resource) {
+ return Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resource);
+ }
+
+ private static PropertyKey getISPropertyKey(HelixDataAccessor accessor, String resource) {
+ return accessor.keyBuilder().idealStates(resource);
+ }
+
+ private static PropertyKey getConfigPropertyKey(HelixDataAccessor accessor, String resource) {
+ return accessor.keyBuilder().resourceConfig(resource);
+ }
+
+ private static void addAllPartitions(Set<Integer> toAdd, Set<Integer> destination) {
+ for (Integer pId : toAdd) {
+ destination.add(pId);
+ }
+ }
+
+ private static ResourceAssignment emptyAssignment(String name) {
+ return new ResourceAssignment(ResourceId.from(name));
+ }
+
+ private static void addCompletedPartitions(Set<Integer> set, JobContext ctx,
+ Iterable<Integer> pIds) {
+ for (Integer pId : pIds) {
+ TaskPartitionState state = ctx.getPartitionState(pId);
+ if (state == TaskPartitionState.COMPLETED) {
+ set.add(pId);
}
}
- return taskPartitions;
+ }
+
+ private static List<Integer> getNextPartitions(SortedSet<Integer> candidatePartitions,
+ Set<Integer> excluded, int n) {
+ List<Integer> result = new ArrayList<Integer>();
+ for (Integer pId : candidatePartitions) {
+ if (result.size() >= n) {
+ break;
+ }
+
+ if (!excluded.contains(pId)) {
+ result.add(pId);
+ }
+ }
+
+ return result;
+ }
+
+ private static void markPartitionCompleted(JobContext ctx, int pId) {
+ ctx.setPartitionState(pId, TaskPartitionState.COMPLETED);
+ ctx.setPartitionFinishTime(pId, System.currentTimeMillis());
+ ctx.incrementNumAttempts(pId);
+ }
+
+ private static void markPartitionError(JobContext ctx, int pId, TaskPartitionState state) {
+ ctx.setPartitionState(pId, state);
+ ctx.setPartitionFinishTime(pId, System.currentTimeMillis());
+ ctx.incrementNumAttempts(pId);
}
/**
- * Get partition assignments for the target resource, but only for the partitions of interest.
- * @param currStateOutput The current state of the instances in the cluster.
- * @param instanceList The set of instances.
- * @param tgtIs The ideal state of the target resource.
- * @param tgtStates Only partitions in this set of states will be considered. If null, partitions
- * do not need to
- * be in any specific state to be considered.
- * @param includeSet The set of partitions to consider.
- * @return A map of instance vs set of partition ids assigned to that instance.
+ * Return the assignment of task partitions per instance.
*/
- private static Map<String, SortedSet<Integer>> getTgtPartitionAssignment(
- ResourceCurrentState currStateOutput, Iterable<ParticipantId> instanceList, IdealState tgtIs,
- Set<String> tgtStates, Set<Integer> includeSet) {
- Map<String, SortedSet<Integer>> result = new HashMap<String, SortedSet<Integer>>();
+ private static Map<ParticipantId, SortedSet<Integer>> getTaskPartitionAssignments(
+ Iterable<ParticipantId> instanceList, ResourceAssignment assignment, Set<Integer> includeSet) {
+ Map<ParticipantId, SortedSet<Integer>> result =
+ new HashMap<ParticipantId, SortedSet<Integer>>();
for (ParticipantId instance : instanceList) {
- result.put(instance.stringify(), new TreeSet<Integer>());
+ result.put(instance, new TreeSet<Integer>());
}
- for (String pName : tgtIs.getPartitionSet()) {
- int pId = pId(pName);
+ for (PartitionId partition : assignment.getMappedPartitionIds()) {
+ int pId = pId(partition.toString());
if (includeSet.contains(pId)) {
- for (ParticipantId instance : instanceList) {
- State s =
- currStateOutput.getCurrentState(ResourceId.from(tgtIs.getResourceName()),
- PartitionId.from(pName), instance);
- String state = (s == null ? null : s.toString());
- if (tgtStates == null || tgtStates.contains(state)) {
- result.get(instance).add(pId);
+ Map<ParticipantId, State> replicaMap = assignment.getReplicaMap(partition);
+ for (ParticipantId instance : replicaMap.keySet()) {
+ SortedSet<Integer> pList = result.get(instance);
+ if (pList != null) {
+ pList.add(pId);
}
}
}
@@ -132,4 +605,32 @@ public class TaskRebalancer extends AbstractTaskRebalancer {
return result;
}
+
+ /**
+ * Computes the partition name given the resource name and partition id.
+ */
+ protected static String pName(String resource, int pId) {
+ return resource + "_" + pId;
+ }
+
+ /**
+ * Extracts the partition id from the given partition name.
+ */
+ protected static int pId(String pName) {
+ String[] tokens = pName.split("_");
+ return Integer.valueOf(tokens[tokens.length - 1]);
+ }
+
+ /**
+ * An (instance, state) pair.
+ */
+ private static class PartitionAssignment {
+ private final String _instance;
+ private final String _state;
+
+ private PartitionAssignment(String instance, String state) {
+ _instance = instance;
+ _state = state;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java b/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
index 92976b0..cd909ed 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
@@ -29,7 +29,6 @@ import org.apache.log4j.Logger;
*/
public class TaskRunner implements Runnable {
private static final Logger LOG = Logger.getLogger(TaskRunner.class);
-
private final StateModel _taskStateModel;
private final HelixManager _manager;
private final String _taskName;
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java b/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
index c399930..a44a8cb 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
@@ -34,12 +34,7 @@ import org.apache.helix.participant.statemachine.StateModelInfo;
import org.apache.helix.participant.statemachine.Transition;
import org.apache.log4j.Logger;
-/**
- * task state model
- */
-@StateModelInfo(states = {
- "INIT", "RUNNING", "STOPPED", "COMPLETED", "TIMED_OUT", "TASK_ERROR", "DROPPED"
-}, initialState = "INIT")
+@StateModelInfo(states = "{'NOT USED BY HELIX'}", initialState = "INIT")
public class TaskStateModel extends StateModel {
private static final Logger LOG = Logger.getLogger(TaskStateModel.class);
private final HelixManager _manager;
@@ -217,19 +212,37 @@ public class TaskStateModel extends StateModel {
}
private void startTask(Message msg, String taskPartition) {
- TaskConfig cfg = TaskUtil.getTaskCfg(_manager, msg.getResourceName());
+ JobConfig cfg = TaskUtil.getJobCfg(_manager, msg.getResourceName());
+ TaskConfig taskConfig = null;
String command = cfg.getCommand();
- Map<String, String> taskNameMap = cfg.getTaskNameMap();
- if (taskNameMap != null && taskNameMap.containsKey(taskPartition)) {
- // Support a partition-specifc override of tasks to run
- String taskName = taskNameMap.get(taskPartition);
- if (_taskFactoryRegistry.containsKey(taskName)) {
- command = taskName;
+
+ // Get a task-specific command if specified
+ JobContext ctx = TaskUtil.getJobContext(_manager, msg.getResourceName());
+ int pId = Integer.parseInt(taskPartition.substring(taskPartition.lastIndexOf('_') + 1));
+ if (ctx.getTaskIdForPartition(pId) != null) {
+ taskConfig = cfg.getTaskConfig(ctx.getTaskIdForPartition(pId));
+ if (taskConfig != null) {
+ if (taskConfig.getCommand() != null) {
+ command = taskConfig.getCommand();
+ }
}
}
+
+ // Populate a task callback context
+ TaskCallbackContext callbackContext = new TaskCallbackContext();
+ callbackContext.setManager(_manager);
+ callbackContext.setJobConfig(cfg);
+ callbackContext.setTaskConfig(taskConfig);
+
+ // Create a task instance with this command
+ if (command == null || _taskFactoryRegistry == null
+ || !_taskFactoryRegistry.containsKey(command)) {
+ throw new IllegalStateException("No callback implemented for task " + command);
+ }
TaskFactory taskFactory = _taskFactoryRegistry.get(command);
- Task task = taskFactory.createNewTask(cfg.getCommandConfig());
+ Task task = taskFactory.createNewTask(callbackContext);
+ // Submit the task for execution
_taskRunner =
new TaskRunner(this, task, msg.getResourceName(), taskPartition, msg.getTgtName(),
_manager, msg.getTgtSessionId());
@@ -244,6 +257,6 @@ public class TaskStateModel extends StateModel {
_taskRunner.timeout();
}
}
- }, cfg.getTimeoutPerPartition());
+ }, cfg.getTimeoutPerTask());
}
}
http://git-wip-us.apache.org/repos/asf/helix/blob/97ca4de4/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 0f980b8..96b7e55 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
@@ -19,11 +19,17 @@ package org.apache.helix.task;
* under the License.
*/
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.helix.AccessOption;
+import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.HelixProperty;
import org.apache.helix.PropertyKey;
import org.apache.helix.ZNRecord;
import org.apache.helix.api.State;
@@ -31,54 +37,48 @@ import org.apache.helix.api.id.PartitionId;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.model.ResourceAssignment;
-import org.apache.helix.model.ResourceConfiguration;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
import org.apache.log4j.Logger;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.type.TypeReference;
import com.google.common.base.Joiner;
+import com.google.common.collect.Maps;
/**
* Static utility methods.
*/
public class TaskUtil {
private static final Logger LOG = Logger.getLogger(TaskUtil.class);
-
- enum TaskUtilEnum {
- CONTEXT_NODE("Context"),
- PREV_RA_NODE("PreviousResourceAssignment");
-
- final String _value;
-
- private TaskUtilEnum(String value) {
- _value = value;
- }
-
- public String value() {
- return _value;
- }
- }
+ private static final String CONTEXT_NODE = "Context";
+ private static final String PREV_RA_NODE = "PreviousResourceAssignment";
/**
- * Parses task resource configurations in Helix into a {@link TaskConfig} object.
+ * Parses job resource configurations in Helix into a {@link JobConfig} object.
* @param manager HelixManager object used to connect to Helix.
- * @param taskResource The name of the task resource.
- * @return A {@link TaskConfig} object if Helix contains valid configurations for the task, null
+ * @param jobResource The name of the job resource.
+ * @return A {@link JobConfig} object if Helix contains valid configurations for the job, null
* otherwise.
*/
- public static TaskConfig getTaskCfg(HelixManager manager, String taskResource) {
- ResourceConfiguration config = getResourceConfig(manager, taskResource);
- Map<String, String> taskCfg = config.getRecord().getSimpleFields();
- TaskConfig.Builder b = TaskConfig.Builder.fromMap(taskCfg);
- if (config.getRecord().getMapFields().containsKey(TaskConfig.TASK_NAME_MAP)) {
- b.setTaskNameMap(config.getRecord().getMapField(TaskConfig.TASK_NAME_MAP));
+ public static JobConfig getJobCfg(HelixManager manager, String jobResource) {
+ HelixProperty jobResourceConfig = getResourceConfig(manager, jobResource);
+ JobConfig.Builder b =
+ JobConfig.Builder.fromMap(jobResourceConfig.getRecord().getSimpleFields());
+ Map<String, Map<String, String>> rawTaskConfigMap =
+ jobResourceConfig.getRecord().getMapFields();
+ Map<String, TaskConfig> taskConfigMap = Maps.newHashMap();
+ for (Map<String, String> rawTaskConfig : rawTaskConfigMap.values()) {
+ TaskConfig taskConfig = TaskConfig.from(rawTaskConfig);
+ taskConfigMap.put(taskConfig.getId(), taskConfig);
}
+ b.addTaskConfigMap(taskConfigMap);
return b.build();
}
public static WorkflowConfig getWorkflowCfg(HelixManager manager, String workflowResource) {
- ResourceConfiguration config = getResourceConfig(manager, workflowResource);
- Map<String, String> workflowCfg = config.getRecord().getSimpleFields();
+ Map<String, String> workflowCfg = getResourceConfigMap(manager, workflowResource);
WorkflowConfig.Builder b = WorkflowConfig.Builder.fromMap(workflowCfg);
+
return b.build();
}
@@ -109,56 +109,97 @@ public class TaskUtil {
String resourceName) {
ZNRecord r =
manager.getHelixPropertyStore().get(
- Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName,
- TaskUtilEnum.PREV_RA_NODE.value()), null, AccessOption.PERSISTENT);
+ Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName, PREV_RA_NODE),
+ null, AccessOption.PERSISTENT);
return r != null ? new ResourceAssignment(r) : null;
}
public static void setPrevResourceAssignment(HelixManager manager, String resourceName,
ResourceAssignment ra) {
manager.getHelixPropertyStore().set(
- Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName,
- TaskUtilEnum.PREV_RA_NODE.value()), ra.getRecord(), AccessOption.PERSISTENT);
+ Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName, PREV_RA_NODE),
+ ra.getRecord(), AccessOption.PERSISTENT);
}
- public static TaskContext getTaskContext(HelixManager manager, String taskResource) {
+ public static JobContext getJobContext(HelixManager manager, String jobResource) {
ZNRecord r =
manager.getHelixPropertyStore().get(
- Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, taskResource,
- TaskUtilEnum.CONTEXT_NODE.value()), null, AccessOption.PERSISTENT);
- return r != null ? new TaskContext(r) : null;
+ Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, jobResource, CONTEXT_NODE),
+ null, AccessOption.PERSISTENT);
+ return r != null ? new JobContext(r) : null;
}
- public static void setTaskContext(HelixManager manager, String taskResource, TaskContext ctx) {
+ public static void setJobContext(HelixManager manager, String jobResource, JobContext ctx) {
manager.getHelixPropertyStore().set(
- Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, taskResource,
- TaskUtilEnum.CONTEXT_NODE.value()), ctx.getRecord(), AccessOption.PERSISTENT);
+ Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, jobResource, CONTEXT_NODE),
+ ctx.getRecord(), AccessOption.PERSISTENT);
}
public static WorkflowContext getWorkflowContext(HelixManager manager, String workflowResource) {
ZNRecord r =
manager.getHelixPropertyStore().get(
Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowResource,
- TaskUtilEnum.CONTEXT_NODE.value()), null, AccessOption.PERSISTENT);
+ CONTEXT_NODE), null, AccessOption.PERSISTENT);
return r != null ? new WorkflowContext(r) : null;
}
public static void setWorkflowContext(HelixManager manager, String workflowResource,
WorkflowContext ctx) {
manager.getHelixPropertyStore().set(
- Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowResource,
- TaskUtilEnum.CONTEXT_NODE.value()), ctx.getRecord(), AccessOption.PERSISTENT);
+ Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowResource, CONTEXT_NODE),
+ ctx.getRecord(), AccessOption.PERSISTENT);
+ }
+
+ public static String getNamespacedJobName(String singleJobWorkflow) {
+ return getNamespacedJobName(singleJobWorkflow, singleJobWorkflow);
}
- public static String getNamespacedTaskName(String singleTaskWorkflow) {
- return getNamespacedTaskName(singleTaskWorkflow, singleTaskWorkflow);
+ public static String getNamespacedJobName(String workflowResource, String jobName) {
+ return workflowResource + "_" + jobName;
}
- public static String getNamespacedTaskName(String workflowResource, String taskName) {
- return workflowResource + "_" + taskName;
+ public static String serializeJobConfigMap(Map<String, String> commandConfig) {
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ String serializedMap = mapper.writeValueAsString(commandConfig);
+ return serializedMap;
+ } catch (IOException e) {
+ LOG.error("Error serializing " + commandConfig, e);
+ }
+ return null;
+ }
+
+ public static Map<String, String> deserializeJobConfigMap(String commandConfig) {
+ ObjectMapper mapper = new ObjectMapper();
+ try {
+ Map<String, String> commandConfigMap =
+ mapper.readValue(commandConfig, new TypeReference<HashMap<String, String>>() {
+ });
+ return commandConfigMap;
+ } catch (IOException e) {
+ LOG.error("Error deserializing " + commandConfig, e);
+ }
+ return Collections.emptyMap();
+ }
+
+ private static Map<String, String> getResourceConfigMap(HelixManager manager, String resource) {
+ HelixConfigScope scope = getResourceConfigScope(manager.getClusterName(), resource);
+ ConfigAccessor configAccessor = manager.getConfigAccessor();
+
+ Map<String, String> taskCfg = new HashMap<String, String>();
+ List<String> cfgKeys = configAccessor.getKeys(scope);
+ if (cfgKeys == null || cfgKeys.isEmpty()) {
+ return null;
+ }
+
+ for (String cfgKey : cfgKeys) {
+ taskCfg.put(cfgKey, configAccessor.get(scope, cfgKey));
+ }
+
+ return getResourceConfig(manager, resource).getRecord().getSimpleFields();
}
- private static ResourceConfiguration getResourceConfig(HelixManager manager, String resource) {
+ private static HelixProperty getResourceConfig(HelixManager manager, String resource) {
HelixDataAccessor accessor = manager.getHelixDataAccessor();
PropertyKey.Builder keyBuilder = accessor.keyBuilder();
return accessor.getProperty(keyBuilder.resourceConfig(resource));