helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From l...@apache.org
Subject [29/33] helix git commit: Populate Error message from running client's task and persist it into JobContext for better error reporting.
Date Wed, 17 Aug 2016 04:27:25 GMT
Populate Error message from running client's task and persist it into JobContext for better
error reporting.


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

Branch: refs/heads/helix-0.6.x
Commit: 9508a1acfae1d915148138daccb2abd5f9dce430
Parents: c3624e0
Author: Lei Xia <lxia@linkedin.com>
Authored: Thu May 5 11:25:22 2016 -0700
Committer: Lei Xia <lxia@linkedin.com>
Committed: Tue Jul 5 16:18:34 2016 -0700

----------------------------------------------------------------------
 .../java/org/apache/helix/task/JobContext.java  |  51 ++++++--
 .../org/apache/helix/task/JobRebalancer.java    |  15 ++-
 .../java/org/apache/helix/task/TaskRunner.java  |   5 +-
 .../org/apache/helix/task/TaskStateModel.java   |   4 +-
 .../apache/helix/integration/task/MockTask.java |  23 +++-
 .../task/TestTaskErrorReporting.java            | 117 +++++++++++++++++++
 6 files changed, 194 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/helix/blob/9508a1ac/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
index 2057f27..328fcc0 100644
--- a/helix-core/src/main/java/org/apache/helix/task/JobContext.java
+++ b/helix-core/src/main/java/org/apache/helix/task/JobContext.java
@@ -45,7 +45,8 @@ public class JobContext extends HelixProperty {
     TARGET,
     TASK_ID,
     ASSIGNED_PARTICIPANT,
-    NEXT_RETRY_TIME
+    NEXT_RETRY_TIME,
+    INFO
   }
 
   public JobContext(ZNRecord record) {
@@ -76,8 +77,18 @@ public class JobContext extends HelixProperty {
     return Long.parseLong(tStr);
   }
 
+  public void setInfo(String info) {
+    if (info != null) {
+      _record.setSimpleField(ContextProperties.INFO.toString(), info);
+    }
+  }
+
+  public String getInfo() {
+    return _record.getSimpleField(ContextProperties.INFO.toString());
+  }
+
   public void setPartitionState(int p, TaskPartitionState s) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.STATE.toString(), s.name());
   }
 
@@ -95,7 +106,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setPartitionNumAttempts(int p, int n) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.NUM_ATTEMPTS.toString(), String.valueOf(n));
   }
 
@@ -122,7 +133,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setPartitionStartTime(int p, long t) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.START_TIME.toString(), String.valueOf(t));
   }
 
@@ -139,7 +150,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setPartitionFinishTime(int p, long t) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.FINISH_TIME.toString(), String.valueOf(t));
   }
 
@@ -156,7 +167,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setPartitionTarget(int p, String targetPName) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.TARGET.toString(), targetPName);
   }
 
@@ -165,6 +176,16 @@ public class JobContext extends HelixProperty {
     return (map != null) ? map.get(ContextProperties.TARGET.toString()) : null;
   }
 
+  public void setPartitionInfo(int p, String info) {
+    Map<String, String> map = getMapField(p, true);
+    map.put(ContextProperties.INFO.toString(), info);
+  }
+
+  public String getPartitionInfo(int p) {
+    Map<String, String> map = getMapField(p);
+    return (map != null) ? map.get(ContextProperties.INFO.toString()) : null;
+  }
+
   public Map<String, List<Integer>> getPartitionsByTarget() {
     Map<String, List<Integer>> result = Maps.newHashMap();
     for (Map.Entry<String, Map<String, String>> mapField : _record.getMapFields().entrySet())
{
@@ -194,7 +215,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setTaskIdForPartition(int p, String taskId) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.TASK_ID.toString(), taskId);
   }
 
@@ -216,7 +237,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setAssignedParticipant(int p, String participantName) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.ASSIGNED_PARTICIPANT.toString(), participantName);
   }
 
@@ -226,7 +247,7 @@ public class JobContext extends HelixProperty {
   }
 
   public void setNextRetryTime(int p, long t) {
-    Map<String, String> map = getMapField(p);
+    Map<String, String> map = getMapField(p, true);
     map.put(ContextProperties.NEXT_RETRY_TIME.toString(), String.valueOf(t));
   }
 
@@ -242,10 +263,20 @@ public class JobContext extends HelixProperty {
     return Long.parseLong(tStr);
   }
 
+  /**
+   * Get MapField for the given partition.
+   *
+   * @param p
+   * @return mapField for the partition, NULL if the partition has not scheduled yet.
+   */
   public Map<String, String> getMapField(int p) {
+    return getMapField(p, false);
+  }
+
+  private Map<String, String> getMapField(int p, boolean createIfNotPresent) {
     String pStr = String.valueOf(p);
     Map<String, String> map = _record.getMapField(pStr);
-    if (map == null) {
+    if (map == null && createIfNotPresent) {
       map = new TreeMap<String, String>();
       _record.setMapField(pStr, map);
     }

http://git-wip-us.apache.org/repos/asf/helix/blob/9508a1ac/helix-core/src/main/java/org/apache/helix/task/JobRebalancer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/JobRebalancer.java b/helix-core/src/main/java/org/apache/helix/task/JobRebalancer.java
index 0f34178..fae7ac7 100644
--- a/helix-core/src/main/java/org/apache/helix/task/JobRebalancer.java
+++ b/helix-core/src/main/java/org/apache/helix/task/JobRebalancer.java
@@ -214,8 +214,9 @@ public class JobRebalancer extends TaskRebalancer {
 
     if (allPartitions == null || allPartitions.isEmpty()) {
       // Empty target partitions, mark the job as FAILED.
-      LOG.warn(
-          "Missing task partition mapping for job " + jobResource + ", marked the job as
FAILED!");
+      String failureMsg = "Empty task partition mapping for job " + jobResource + ", marked
the job as FAILED!";
+      LOG.info(failureMsg);
+      jobCtx.setInfo(failureMsg);
       markJobFailed(jobResource, jobCtx, workflowConfig, workflowCtx);
       markAllPartitionsError(jobCtx, TaskPartitionState.ERROR, false);
       return new ResourceAssignment(jobResource);
@@ -266,6 +267,12 @@ public class JobRebalancer extends TaskRebalancer {
                 pName), instance));
         jobCtx.setPartitionState(pId, currState);
 
+        String taskMsg = currStateOutput.getInfo(jobResource, new Partition(
+            pName), instance);
+        if (taskMsg != null) {
+          jobCtx.setPartitionInfo(pId, taskMsg);
+        }
+
         // Process any requested state transitions.
         String requestedStateStr =
             currStateOutput.getRequestedState(jobResource, new Partition(pName), instance);
@@ -318,8 +325,8 @@ public class JobRebalancer extends TaskRebalancer {
         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));
+              "Task partition %s has error state %s with msg %s. Marking as such in rebalancer
context.", pName,
+              currState, taskMsg));
           markPartitionError(jobCtx, pId, currState, true);
           // The error policy is to fail the task as soon a single partition fails for a
specified
           // maximum number of attempts or task is in ABORTED state.

http://git-wip-us.apache.org/repos/asf/helix/blob/9508a1ac/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 c43d0ce..eabaf64 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
@@ -72,7 +72,7 @@ public class TaskRunner implements Runnable {
         throw death;
       } catch (Throwable t) {
         LOG.error("Problem running the task, report task as FAILED.", t);
-        _result = new TaskResult(Status.FAILED, null);
+        _result = new TaskResult(Status.FAILED, "Exception happened in running task: " +
t.getMessage());
       }
 
       switch (_result.getStatus()) {
@@ -98,6 +98,9 @@ public class TaskRunner implements Runnable {
         throw new AssertionError("Unknown task result type: " + _result.getStatus().name());
       }
     } catch (Exception e) {
+      LOG.error("Problem running the task, report task as FAILED.", e);
+      _result =
+          new TaskResult(Status.FAILED, "Exception happened in running task: " + e.getMessage());
       requestStateTransition(TaskPartitionState.TASK_ERROR);
     } finally {
       synchronized (_doneSync) {

http://git-wip-us.apache.org/repos/asf/helix/blob/9508a1ac/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 ba68a78..fd07176 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
@@ -87,7 +87,7 @@ public class TaskStateModel extends StateModel {
   }
 
   @Transition(to = "COMPLETED", from = "RUNNING")
-  public void onBecomeCompletedFromRunning(Message msg, NotificationContext context) {
+  public String onBecomeCompletedFromRunning(Message msg, NotificationContext context) {
     String taskPartition = msg.getPartitionName();
     if (_taskRunner == null) {
       throw new IllegalStateException(String.format(
@@ -102,6 +102,8 @@ public class TaskStateModel extends StateModel {
     }
 
     timeout_task.cancel(false);
+
+    return r.getInfo();
   }
 
   @Transition(to = "TIMED_OUT", from = "RUNNING")

http://git-wip-us.apache.org/repos/asf/helix/blob/9508a1ac/helix-core/src/test/java/org/apache/helix/integration/task/MockTask.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/MockTask.java b/helix-core/src/test/java/org/apache/helix/integration/task/MockTask.java
index 3fe1d6f..db0c8f4 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/MockTask.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/MockTask.java
@@ -32,12 +32,14 @@ public class MockTask implements Task {
   public static final String TIMEOUT_CONFIG = "Timeout";
   public static final String TASK_RESULT_STATUS = "TaskResultStatus";
   public static final String THROW_EXCEPTION = "ThrowException";
+  public static final String ERROR_MESSAGE = "ErrorMessage";
   public static final String FAILURE_COUNT_BEFORE_SUCCESS = "FailureCountBeforeSuccess";
   private final long _delay;
   private volatile boolean _canceled;
   private TaskResult.Status _taskResultStatus;
   private boolean _throwException;
   private int _expectedToSuccess;
+  private String _errorMsg;
 
   public MockTask(TaskCallbackContext context) {
     Map<String, String> cfg = context.getJobConfig().getJobCommandConfigMap();
@@ -59,8 +61,10 @@ public class MockTask implements Task {
         Boolean.valueOf(cfg.containsKey(THROW_EXCEPTION)) :
         false;
     _expectedToSuccess =
-        cfg.containsKey(FAILURE_COUNT_BEFORE_SUCCESS) ? Integer.parseInt(cfg.get(
-            FAILURE_COUNT_BEFORE_SUCCESS)) : 0;
+        cfg.containsKey(FAILURE_COUNT_BEFORE_SUCCESS) ? Integer.parseInt(
+            cfg.get(FAILURE_COUNT_BEFORE_SUCCESS)) : 0;
+
+    _errorMsg = cfg.containsKey(ERROR_MESSAGE) ? cfg.get(ERROR_MESSAGE) : null;
   }
 
   @Override
@@ -77,12 +81,21 @@ public class MockTask implements Task {
     }
     timeLeft = expiry - System.currentTimeMillis();
 
-    if (_throwException || _expectedToSuccess > 0) {
+    if (_throwException) {
+      _expectedToSuccess--;
+      if (_errorMsg == null) {
+        _errorMsg = "Test failed";
+      }
+      throw new RuntimeException(_errorMsg != null ? _errorMsg : "Test failed");
+    }
+
+    if (_expectedToSuccess > 0){
       _expectedToSuccess--;
-      throw new RuntimeException("Test failed");
+      throw new RuntimeException(_errorMsg != null ? _errorMsg : "Test failed");
     }
 
-    return new TaskResult(_taskResultStatus, String.valueOf(timeLeft < 0 ? 0 : timeLeft));
+    return new TaskResult(_taskResultStatus,
+        _errorMsg != null ? _errorMsg : String.valueOf(timeLeft < 0 ? 0 : timeLeft));
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/helix/blob/9508a1ac/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskErrorReporting.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskErrorReporting.java
b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskErrorReporting.java
new file mode 100644
index 0000000..906dcff
--- /dev/null
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskErrorReporting.java
@@ -0,0 +1,117 @@
+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 org.apache.helix.TestHelper;
+import org.apache.helix.task.JobConfig;
+import org.apache.helix.task.JobContext;
+import org.apache.helix.task.TaskConfig;
+import org.apache.helix.task.TaskPartitionState;
+import org.apache.helix.task.TaskResult;
+import org.apache.helix.task.TaskState;
+import org.apache.helix.task.TaskUtil;
+import org.apache.helix.task.Workflow;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test Error reporting for failed tasks
+ */
+public class TestTaskErrorReporting extends TaskTestBase {
+
+  @Test
+  public void test() throws Exception {
+    int taskRetryCount = 1;
+    int num_tasks = 5;
+
+    String jobResource = TestHelper.getTestMethodName();
+    JobConfig.Builder jobBuilder = new JobConfig.Builder();
+    jobBuilder.setCommand(MockTask.TASK_COMMAND).setTimeoutPerTask(10000)
+        .setMaxAttemptsPerTask(taskRetryCount).setFailureThreshold(Integer.MAX_VALUE);
+
+    // create each task configs.
+    final int abortedTask = 1;
+    final int failedTask = 2;
+    final int exceptionTask = 3;
+
+    final String abortedMsg = "This task aborted, some terrible things must happened.";
+    final String failedMsg = "This task failed, something may be wrong.";
+    final String exceptionMsg = "This task throws exception.";
+    final String successMsg = "Yes, we did it!";
+
+    List<TaskConfig> taskConfigs = new ArrayList<TaskConfig>();
+    for (int j = 0; j < num_tasks; j++) {
+      TaskConfig.Builder configBuilder = new TaskConfig.Builder().setTaskId("task_" + j);
+      switch (j) {
+      case abortedTask:
+        configBuilder.addConfig(MockTask.TASK_RESULT_STATUS, TaskResult.Status.FATAL_FAILED.name())
+            .addConfig(MockTask.ERROR_MESSAGE, abortedMsg);
+        break;
+      case failedTask:
+        configBuilder.addConfig(MockTask.TASK_RESULT_STATUS, TaskResult.Status.FAILED.name())
+            .addConfig(MockTask.ERROR_MESSAGE, failedMsg);
+        break;
+      case exceptionTask:
+        configBuilder.addConfig(MockTask.THROW_EXCEPTION, Boolean.TRUE.toString())
+            .addConfig(MockTask.ERROR_MESSAGE, exceptionMsg);
+        break;
+      default:
+        configBuilder.addConfig(MockTask.ERROR_MESSAGE, successMsg);
+        break;
+      }
+      configBuilder.setTargetPartition(String.valueOf(j));
+      taskConfigs.add(configBuilder.build());
+    }
+    jobBuilder.addTaskConfigs(taskConfigs);
+
+    Workflow flow =
+        WorkflowGenerator.generateSingleJobWorkflowBuilder(jobResource, jobBuilder).build();
+
+    _driver.start(flow);
+
+    // Wait until the job completes.
+    TaskTestUtil.pollForWorkflowState(_driver, jobResource, TaskState.COMPLETED);
+
+    JobContext ctx = _driver.getJobContext(TaskUtil.getNamespacedJobName(jobResource));
+    for (int i = 0; i < num_tasks; i++) {
+      TaskPartitionState state = ctx.getPartitionState(i);
+      String taskId = ctx.getTaskIdForPartition(i);
+      String errMsg = ctx.getPartitionInfo(i);
+
+      if (taskId.equals("task_" + abortedTask)) {
+        Assert.assertEquals(state, TaskPartitionState.TASK_ABORTED);
+        Assert.assertEquals(errMsg, abortedMsg);
+      } else if (taskId.equals("task_" + failedTask)) {
+        Assert.assertEquals(state, TaskPartitionState.TASK_ERROR);
+        Assert.assertEquals(errMsg, failedMsg);
+      } else if (taskId.equals("task_" + exceptionTask)) {
+        Assert.assertEquals(state, TaskPartitionState.TASK_ERROR);
+        Assert.assertTrue(errMsg.contains(exceptionMsg));
+      } else {
+        Assert.assertEquals(state, TaskPartitionState.COMPLETED);
+        Assert.assertEquals(errMsg, successMsg);
+
+      }
+    }
+  }
+}


Mime
View raw message