helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ka...@apache.org
Subject [43/50] [abbrv] git commit: [HELIX-439] Support thresholding for job success/failure
Date Thu, 10 Jul 2014 17:05:26 GMT
[HELIX-439] Support thresholding for job success/failure


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

Branch: refs/heads/master
Commit: c5921f4299cd3d6d6be1aed44053904e1aada53e
Parents: feaea56
Author: Kanak Biscuitwala <kanak@apache.org>
Authored: Tue Jul 8 18:34:57 2014 -0700
Committer: Kanak Biscuitwala <kanak@apache.org>
Committed: Tue Jul 8 18:34:57 2014 -0700

----------------------------------------------------------------------
 .../java/org/apache/helix/task/JobConfig.java   | 27 ++++++-
 .../java/org/apache/helix/task/TaskConfig.java  | 36 +++++++--
 .../org/apache/helix/task/TaskRebalancer.java   | 52 +++++++++----
 .../java/org/apache/helix/task/Workflow.java    |  2 +
 .../org/apache/helix/task/beans/JobBean.java    |  1 +
 .../org/apache/helix/task/beans/TaskBean.java   |  1 +
 .../task/TestIndependentTaskRebalancer.java     | 81 +++++++++++++++++++-
 7 files changed, 174 insertions(+), 26 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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
index 90e3cfc..b166da1 100644
--- a/helix-core/src/main/java/org/apache/helix/task/JobConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/task/JobConfig.java
@@ -63,6 +63,8 @@ public class JobConfig {
   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 number of tasks within the job that are allowed to fail. */
+  public static final String FAILURE_THRESHOLD = "FailureThreshold";
 
   /** The individual task configurations, if any **/
   public static final String TASK_CONFIGS = "TaskConfigs";
@@ -72,6 +74,7 @@ public class JobConfig {
   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;
+  public static final int DEFAULT_FAILURE_THRESHOLD = 0;
 
   private final String _workflow;
   private final String _targetResource;
@@ -82,12 +85,13 @@ public class JobConfig {
   private final long _timeoutPerTask;
   private final int _numConcurrentTasksPerInstance;
   private final int _maxAttemptsPerTask;
+  private final int _failureThreshold;
   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) {
+      int failureThreshold, Map<String, TaskConfig> taskConfigMap) {
     _workflow = workflow;
     _targetResource = targetResource;
     _targetPartitions = targetPartitions;
@@ -97,6 +101,7 @@ public class JobConfig {
     _timeoutPerTask = timeoutPerTask;
     _numConcurrentTasksPerInstance = numConcurrentTasksPerInstance;
     _maxAttemptsPerTask = maxAttemptsPerTask;
+    _failureThreshold = failureThreshold;
     if (taskConfigMap != null) {
       _taskConfigMap = taskConfigMap;
     } else {
@@ -140,6 +145,10 @@ public class JobConfig {
     return _maxAttemptsPerTask;
   }
 
+  public int getFailureThreshold() {
+    return _failureThreshold;
+  }
+
   public Map<String, TaskConfig> getTaskConfigMap() {
     return _taskConfigMap;
   }
@@ -171,6 +180,7 @@ public class JobConfig {
     }
     cfgMap.put(JobConfig.TIMEOUT_PER_TASK, "" + _timeoutPerTask);
     cfgMap.put(JobConfig.MAX_ATTEMPTS_PER_TASK, "" + _maxAttemptsPerTask);
+    cfgMap.put(JobConfig.FAILURE_THRESHOLD, "" + _failureThreshold);
     return cfgMap;
   }
 
@@ -188,13 +198,14 @@ public class JobConfig {
     private long _timeoutPerTask = DEFAULT_TIMEOUT_PER_TASK;
     private int _numConcurrentTasksPerInstance = DEFAULT_NUM_CONCURRENT_TASKS_PER_INSTANCE;
     private int _maxAttemptsPerTask = DEFAULT_MAX_ATTEMPTS_PER_TASK;
+    private int _failureThreshold = DEFAULT_FAILURE_THRESHOLD;
 
     public JobConfig build() {
       validate();
 
       return new JobConfig(_workflow, _targetResource, _targetPartitions, _targetPartitionStates,
           _command, _commandConfig, _timeoutPerTask, _numConcurrentTasksPerInstance,
-          _maxAttemptsPerTask, _taskConfigMap);
+          _maxAttemptsPerTask, _failureThreshold, _taskConfigMap);
     }
 
     /**
@@ -235,6 +246,9 @@ public class JobConfig {
       if (cfg.containsKey(MAX_ATTEMPTS_PER_TASK)) {
         b.setMaxAttemptsPerTask(Integer.parseInt(cfg.get(MAX_ATTEMPTS_PER_TASK)));
       }
+      if (cfg.containsKey(FAILURE_THRESHOLD)) {
+        b.setFailureThreshold(Integer.parseInt(cfg.get(FAILURE_THRESHOLD)));
+      }
       return b;
     }
 
@@ -283,6 +297,11 @@ public class JobConfig {
       return this;
     }
 
+    public Builder setFailureThreshold(int v) {
+      _failureThreshold = v;
+      return this;
+    }
+
     public Builder addTaskConfigs(List<TaskConfig> taskConfigs) {
       if (taskConfigs != null) {
         for (TaskConfig taskConfig : taskConfigs) {
@@ -321,6 +340,10 @@ public class JobConfig {
         throw new IllegalArgumentException(String.format("%s has invalid value %s",
             MAX_ATTEMPTS_PER_TASK, _maxAttemptsPerTask));
       }
+      if (_failureThreshold < 0) {
+        throw new IllegalArgumentException(String.format("%s has invalid value %s",
+            FAILURE_THRESHOLD, _failureThreshold));
+      }
       if (_workflow == null) {
         throw new IllegalArgumentException(String.format("%s cannot be null", WORKFLOW_ID));
       }

http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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 547ba48..4ddab1a 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
@@ -35,7 +35,8 @@ import com.google.common.collect.Maps;
 public class TaskConfig {
   private enum TaskConfigFields {
     TASK_ID,
-    TASK_COMMAND
+    TASK_COMMAND,
+    TASK_SUCCESS_OPTIONAL
   }
 
   private static final Logger LOG = Logger.getLogger(TaskConfig.class);
@@ -46,9 +47,12 @@ public class TaskConfig {
    * 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 successOptional true if this task need not pass for the job to succeed, false
+   *          otherwise
    * @param id existing task ID
    */
-  public TaskConfig(String command, Map<String, String> configMap, String id) {
+  public TaskConfig(String command, Map<String, String> configMap, boolean successOptional,
+      String id) {
     if (configMap == null) {
       configMap = Maps.newHashMap();
     }
@@ -56,6 +60,8 @@ public class TaskConfig {
       id = UUID.randomUUID().toString();
     }
     configMap.put(TaskConfigFields.TASK_COMMAND.toString(), command);
+    configMap.put(TaskConfigFields.TASK_SUCCESS_OPTIONAL.toString(),
+        Boolean.toString(successOptional));
     configMap.put(TaskConfigFields.TASK_ID.toString(), id);
     _configMap = configMap;
   }
@@ -64,9 +70,11 @@ public class TaskConfig {
    * 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 successOptional true if this task need not pass for the job to succeed, false
+   *          otherwise
    */
-  public TaskConfig(String command, Map<String, String> configMap) {
-    this(command, configMap, null);
+  public TaskConfig(String command, Map<String, String> configMap, boolean successOptional)
{
+    this(command, configMap, successOptional, null);
   }
 
   /**
@@ -86,6 +94,19 @@ public class TaskConfig {
   }
 
   /**
+   * Check if this task must succeed for a job to succeed
+   * @return true if success is optional, false otherwise
+   */
+  public boolean isSuccessOptional() {
+    String successOptionalStr = _configMap.get(TaskConfigFields.TASK_SUCCESS_OPTIONAL.toString());
+    if (successOptionalStr == null) {
+      return false;
+    } else {
+      return Boolean.parseBoolean(successOptionalStr);
+    }
+  }
+
+  /**
    * Get the configuration map for this task's command
    * @return map of configuration key to value
    */
@@ -110,7 +131,7 @@ public class TaskConfig {
    * @return instantiated TaskConfig
    */
   public static TaskConfig from(TaskBean bean) {
-    return new TaskConfig(bean.command, bean.taskConfigMap);
+    return new TaskConfig(bean.command, bean.taskConfigMap, bean.successOptional);
   }
 
   /**
@@ -121,6 +142,9 @@ public class 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);
+    String successOptionalStr = rawConfigMap.get(TaskConfigFields.TASK_SUCCESS_OPTIONAL.toString());
+    boolean successOptional =
+        (successOptionalStr != null) ? Boolean.valueOf(successOptionalStr) : null;
+    return new TaskConfig(command, rawConfigMap, successOptional, taskId);
   }
 }

http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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 e9f60f9..376eca5 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
@@ -213,6 +213,9 @@ public abstract class TaskRebalancer implements HelixRebalancer {
     // Used to keep track of tasks that have already been assigned to instances.
     Set<Integer> assignedPartitions = new HashSet<Integer>();
 
+    // Used to keep track of tasks that have failed, but whose failure is acceptable
+    Set<Integer> skippedPartitions = new HashSet<Integer>();
+
     // Keeps a mapping of (partition) -> (instance, state)
     Map<Integer, PartitionAssignment> paMap = new TreeMap<Integer, PartitionAssignment>();
 
@@ -227,7 +230,6 @@ public abstract class TaskRebalancer implements HelixRebalancer {
       // TASK_ERROR, ERROR.
       Set<Integer> donePartitions = new TreeSet<Integer>();
       for (int pId : pSet) {
-        jobCtx.setPartitionState(pId, TaskPartitionState.INIT);
         final String pName = pName(jobResource, pId);
 
         // Check for pending state transitions on this (partition, instance).
@@ -236,8 +238,7 @@ public abstract class TaskRebalancer implements HelixRebalancer {
                 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.
+          // the state assignment from the previous ideal state.
           Map<ParticipantId, State> stateMap =
               prevAssignment.getReplicaMap(PartitionId.from(pName));
           if (stateMap != null) {
@@ -290,8 +291,6 @@ public abstract class TaskRebalancer implements HelixRebalancer {
             nextState = TaskPartitionState.STOPPED;
           }
 
-          jobCtx.setPartitionState(pId, currState);
-
           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,
@@ -318,13 +317,34 @@ public abstract class TaskRebalancer implements HelixRebalancer {
               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.
+          // 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);
+            // If the user does not require this task to succeed in order for the job to
succeed,
+            // then we don't have to fail the job right now
+            boolean successOptional = false;
+            String taskId = jobCtx.getTaskIdForPartition(pId);
+            if (taskId != null) {
+              TaskConfig taskConfig = jobCfg.getTaskConfig(taskId);
+              if (taskConfig != null) {
+                successOptional = taskConfig.isSuccessOptional();
+              }
+            }
+
+            // Similarly, if we have some leeway for how many tasks we can fail, then we
don't have
+            // to fail the job immediately
+            if (skippedPartitions.size() < jobCfg.getFailureThreshold()) {
+              successOptional = true;
+            }
+
+            if (!successOptional) {
+              workflowCtx.setJobState(jobResource, TaskState.FAILED);
+              workflowCtx.setWorkflowState(TaskState.FAILED);
+              addAllPartitions(allPartitions, partitionsToDropFromIs);
+              return emptyAssignment(jobResource);
+            } else {
+              skippedPartitions.add(pId);
+              partitionsToDropFromIs.add(pId);
+            }
           }
         }
           break;
@@ -346,7 +366,7 @@ public abstract class TaskRebalancer implements HelixRebalancer {
       pSet.removeAll(donePartitions);
     }
 
-    if (isJobComplete(jobCtx, allPartitions)) {
+    if (isJobComplete(jobCtx, allPartitions, skippedPartitions)) {
       workflowCtx.setJobState(jobResource, TaskState.COMPLETED);
       if (isWorkflowComplete(workflowCtx, workflowConfig)) {
         workflowCtx.setWorkflowState(TaskState.COMPLETED);
@@ -381,7 +401,6 @@ public abstract class TaskRebalancer implements HelixRebalancer {
             paMap.put(pId,
                 new PartitionAssignment(instance.toString(), TaskPartitionState.RUNNING.name()));
             excludeSet.add(pId);
-            jobCtx.setPartitionState(pId, TaskPartitionState.INIT);
             jobCtx.setAssignedParticipant(pId, instance.toString());
             LOG.debug(String.format("Setting task partition %s state to %s on instance %s.",
pName,
                 TaskPartitionState.RUNNING, instance));
@@ -397,7 +416,6 @@ public abstract class TaskRebalancer implements HelixRebalancer {
       ra.addReplicaMap(PartitionId.from(pName(jobResource, e.getKey())),
           ImmutableMap.of(ParticipantId.from(pa._instance), State.from(pa._state)));
     }
-
     return ra;
   }
 
@@ -405,14 +423,16 @@ public abstract class TaskRebalancer implements HelixRebalancer {
    * Checks if the job has completed.
    * @param ctx The rebalancer context.
    * @param allPartitions The set of partitions to check.
+   * @param skippedPartitions partitions that failed, but whose failure is acceptable
    * @return true if all task partitions have been marked with status
    *         {@link TaskPartitionState#COMPLETED} in the rebalancer
    *         context, false otherwise.
    */
-  private static boolean isJobComplete(JobContext ctx, Set<Integer> allPartitions)
{
+  private static boolean isJobComplete(JobContext ctx, Set<Integer> allPartitions,
+      Set<Integer> skippedPartitions) {
     for (Integer pId : allPartitions) {
       TaskPartitionState state = ctx.getPartitionState(pId);
-      if (state != TaskPartitionState.COMPLETED) {
+      if (!skippedPartitions.contains(pId) && state != TaskPartitionState.COMPLETED)
{
         return false;
       }
     }

http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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 1a41e06..8afafe4 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
@@ -185,6 +185,8 @@ public class Workflow {
           String.valueOf(job.numConcurrentTasksPerInstance));
       builder.addConfig(job.name, JobConfig.TIMEOUT_PER_TASK,
           String.valueOf(job.timeoutPerPartition));
+      builder
+          .addConfig(job.name, JobConfig.FAILURE_THRESHOLD, String.valueOf(job.failureThreshold));
       if (job.tasks != null) {
         List<TaskConfig> taskConfigs = Lists.newArrayList();
         for (TaskBean task : job.tasks) {

http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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
index 5e12f19..af5882c 100644
--- 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
@@ -39,4 +39,5 @@ public class JobBean {
   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;
+  public int failureThreshold = JobConfig.DEFAULT_FAILURE_THRESHOLD;
 }

http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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 eedccb5..97ecfc0 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
@@ -29,4 +29,5 @@ import java.util.Map;
 public class TaskBean {
   public String command;
   public Map<String, String> taskConfigMap;
+  public boolean successOptional = false;
 }

http://git-wip-us.apache.org/repos/asf/helix/blob/c5921f42/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
index 1ee3991..5dad94c 100644
--- 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
@@ -41,6 +41,7 @@ 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.TaskResult.Status;
 import org.apache.helix.task.TaskState;
 import org.apache.helix.task.TaskStateModelFactory;
 import org.apache.helix.task.Workflow;
@@ -51,6 +52,7 @@ import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 import org.testng.collections.Sets;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
@@ -131,8 +133,63 @@ public class TestIndependentTaskRebalancer extends ZkIntegrationTestBase
{
     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);
+    TaskConfig taskConfig1 = new TaskConfig("TaskOne", null, true);
+    TaskConfig taskConfig2 = new TaskConfig("TaskTwo", null, true);
+    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()));
+  }
+
+  @Test
+  public void testThresholdFailure() 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);
+    Map<String, String> taskConfigMap = Maps.newHashMap(ImmutableMap.of("fail", ""
+ true));
+    TaskConfig taskConfig1 = new TaskConfig("TaskOne", taskConfigMap, false);
+    TaskConfig taskConfig2 = new TaskConfig("TaskTwo", null, false);
+    taskConfigs.add(taskConfig1);
+    taskConfigs.add(taskConfig2);
+    workflowBuilder.addTaskConfigs(jobName, taskConfigs);
+    workflowBuilder.addConfig(jobName, JobConfig.COMMAND, "DummyCommand");
+    workflowBuilder.addConfig(jobName, JobConfig.FAILURE_THRESHOLD, "" + 1);
+    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()));
+  }
+
+  @Test
+  public void testOptionalTaskFailure() 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);
+    Map<String, String> taskConfigMap = Maps.newHashMap(ImmutableMap.of("fail", ""
+ true));
+    TaskConfig taskConfig1 = new TaskConfig("TaskOne", taskConfigMap, true);
+    TaskConfig taskConfig2 = new TaskConfig("TaskTwo", null, false);
     taskConfigs.add(taskConfig1);
     taskConfigs.add(taskConfig2);
     workflowBuilder.addTaskConfigs(jobName, taskConfigs);
@@ -152,13 +209,33 @@ public class TestIndependentTaskRebalancer extends ZkIntegrationTestBase
{
   }
 
   private class TaskOne extends ReindexTask {
+    private final boolean _shouldFail;
+
     public TaskOne(TaskCallbackContext context) {
       super(context);
+
+      // Check whether or not this task should succeed
+      TaskConfig taskConfig = context.getTaskConfig();
+      boolean shouldFail = false;
+      if (taskConfig != null) {
+        Map<String, String> configMap = taskConfig.getConfigMap();
+        if (configMap != null && configMap.containsKey("fail")
+            && Boolean.parseBoolean(configMap.get("fail"))) {
+          shouldFail = true;
+        }
+      }
+      _shouldFail = shouldFail;
     }
 
     @Override
     public TaskResult run() {
       _invokedClasses.add(getClass().getName());
+
+      // Fail the task if it should fail
+      if (_shouldFail) {
+        return new TaskResult(Status.ERROR, null);
+      }
+
       return super.run();
     }
   }


Mime
View raw message