Surface per-task status summary in Helix JobContext

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

@@ -507,6 +507,9 @@ protected void handleJobTimeout(JobContext jobCtx, WorkflowContext workflowCtx,
     _clusterStatusMonitor.updateJobCounters(jobCfg, TaskState.TIMED_OUT);
     _rebalanceScheduler.removeScheduledRebalance(jobResource);
     TaskUtil.cleanupJobIdealStateExtView(_manager.getHelixDataAccessor(), jobResource);
+    // Aggregate the terminal state of every task into a job-level summary (computed after INIT
+    // tasks are marked TASK_ABORTED above so the counts reflect the final states).
+    jobCtx.updateTaskStatusSummary();
     // New pipeline trigger for workflow status update
     // TODO: Enhance the pipeline and remove this because this operation is expansive
     RebalanceUtil.scheduleOnDemandPipeline(_manager.getClusterName(),0L,false);
@@ -529,6 +532,10 @@ protected void failJob(String jobName, WorkflowContext workflowContext, JobConte
     // New pipeline trigger for workflow status update
     // TODO: Enhance the pipeline and remove this because this operation is expansive
     RebalanceUtil.scheduleOnDemandPipeline(_manager.getClusterName(),0L,false);
+
+    // Aggregate the terminal state of every task into a job-level summary (computed after INIT
+    // tasks are marked TASK_ABORTED above so the counts reflect the final states).
+    jobContext.updateTaskStatusSummary();
   }
 
   // Compute real assignment from theoretical calculation with applied throttling
@@ -935,6 +942,10 @@ protected void markJobComplete(final String jobName, final JobContext jobContext
       reportControllerInducedDelay(dataProvider, _clusterStatusMonitor, workflowConfig, jobConfig,
           currentTime);
     }
+
+    // Aggregate the terminal state of every task into a job-level summary so that partial failures
+    // remain visible even though the job itself is COMPLETED (e.g. FailureThreshold set high).
+    jobContext.updateTaskStatusSummary();
   }
 
   protected void markJobFailed(String jobName, JobContext jobContext, WorkflowConfig workflowConfig,

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

@@ -20,22 +20,48 @@
  */
 
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import org.apache.helix.HelixProperty;
 import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Provides a typed interface to the context information stored by {@link TaskRebalancer} in the
  * Helix property store.
  */
 public class JobContext extends HelixProperty {
+  private static final Logger LOG = LoggerFactory.getLogger(JobContext.class);
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  // Field names used in the aggregated task status summary JSON produced by
+  // updateTaskStatusSummary(). Kept as named constants (rather than inline string literals) so the
+  // producer here and every consumer that mirrors this shape stay in agreement on the exact keys.
+  static final String SUMMARY_KEY_TOTAL = "total";
+  static final String SUMMARY_KEY_COMPLETED = "completed";
+  static final String SUMMARY_KEY_FAILED = "failed";
+  static final String SUMMARY_KEY_TIMED_OUT = "timedOut";
+  static final String SUMMARY_KEY_IN_PROGRESS = "inProgress";
+  static final String SUMMARY_KEY_PENDING = "pending";
+  static final String SUMMARY_KEY_OTHER = "other";
+  static final String SUMMARY_KEY_BY_STATE = "byState";
+  static final String SUMMARY_KEY_FAILED_TASKS = "failedTasks";
+  static final String SUMMARY_KEY_TIMED_OUT_TASKS = "timedOutTasks";
+  static final String SUMMARY_KEY_IN_PROGRESS_TASKS = "inProgressTasks";
+  static final String SUMMARY_KEY_PENDING_TASKS = "pendingTasks";
+  // byState bucket used for a partition that has no task state yet (getPartitionState == null).
+  static final String SUMMARY_STATE_UNSCHEDULED = "UNSCHEDULED";
+
   private enum ContextProperties {
     START_TIME, // Time at which this JobContext was created
     STATE,
@@ -48,6 +74,9 @@ private enum ContextProperties {
     INFO,
     NAME,
     EXECUTION_START_TIME, // Time at which the first task of this job got scheduled
+    TASK_STATUS_SUMMARY, // Aggregated per-task terminal status summary (JSON), set when the job
+                         // reaches a terminal state. Surfaces partial failures even when the job
+                         // itself is COMPLETED (e.g. FailureThreshold set high so all tasks run).
   }
 
   // Note: This field needs to be set if any of the job context fields have been changed.
@@ -337,6 +366,131 @@ public long getExecutionStartTime() {
     return Long.parseLong(tStr);
   }
 
+  /**
+   * Task partition states that represent a terminal failure (the task was given up and will not be
+   * retried, or it errored/timed out). Used to compute the aggregated task status summary.
+   */
+  private static boolean isFailedState(TaskPartitionState state) {
+    return state == TaskPartitionState.TASK_ERROR || state == TaskPartitionState.TASK_ABORTED
+        || state == TaskPartitionState.TIMED_OUT || state == TaskPartitionState.ERROR;
+  }
+
+  /**
+   * Task partition state for a task that is actively running (RUNNING). Surfaced explicitly instead
+   * of being lumped into "other" so still-running tasks are visible, which matters because the
+   * summary is refreshed live while the job runs, not only when it reaches a terminal state.
+   */
+  private static boolean isInProgressState(TaskPartitionState state) {
+    return state == TaskPartitionState.RUNNING;
+  }
+
+  /**
+   * Task partition state for a task that has been scheduled but has not started running yet (INIT).
+   * Surfaced as its own "pending" count, distinct from "inProgress" (RUNNING), so operators can
+   * tell tasks that are waiting to start apart from tasks that are actively executing.
+   */
+  private static boolean isPendingState(TaskPartitionState state) {
+    return state == TaskPartitionState.INIT;
+  }
+
+  /**
+   * Aggregates the state of every scheduled task in this job into a compact, job-level summary and
+   * stores it as a simple field (JSON). This lets operators observe partial failures even when the
+   * job's own status flag is COMPLETED, which happens when FailureThreshold is set high so that all
+   * partition tasks are allowed to run to completion. Intended to be called by the controller when
+   * the job reaches a terminal state (completed / failed / timed out).
+   * The summary shape is:
+   * {@code {"total":N,"completed":X,"failed":Y,"timedOut":T,"inProgress":R,"pending":P,"other":Z,
+   * "byState":{...},"failedTasks":[...],"timedOutTasks":[...],"inProgressTasks":[...],
+   * "pendingTasks":[...]}}.
+   * The top-level counts partition the tasks as {@code total = completed + failed + inProgress +
+   * pending + other}. {@code failed} is the count of tasks that did not succeed (errored, aborted,
+   * or timed out); {@code timedOut} is the subset of {@code failed} that specifically timed out,
+   * called out separately so operators can distinguish a task that ran out of time from one that
+   * errored or aborted. {@code inProgress} is the count of tasks actively running (RUNNING) and
+   * {@code pending} is the count of tasks scheduled but not yet started (INIT); both are pulled out
+   * of {@code other} so still-active tasks are visible when the summary is recomputed on demand
+   * (for example when the job detail page is opened) while the job is still running.
+   * {@code byState} is a histogram mapping each raw task partition state name (or
+   * {@code "UNSCHEDULED"} for a partition that has not been assigned a state yet) to the number of
+   * tasks in that state; it is the fine-grained breakdown behind the coarser counts above. The
+   * {@code failedTasks} / {@code timedOutTasks} / {@code inProgressTasks} / {@code pendingTasks}
+   * arrays list the partition ids that fall into each of those buckets.
+   */
+  public void updateTaskStatusSummary() {
+    Set<Integer> partitions = getPartitionSet();
+    Map<String, Integer> byState = new TreeMap<>();
+    List<Integer> failedTasks = Lists.newArrayList();
+    List<Integer> timedOutTasks = Lists.newArrayList();
+    List<Integer> inProgressTasks = Lists.newArrayList();
+    List<Integer> pendingTasks = Lists.newArrayList();
+    int completed = 0;
+    int failed = 0;
+    int timedOut = 0;
+    int inProgress = 0;
+    int pending = 0;
+    for (int p : partitions) {
+      TaskPartitionState state = getPartitionState(p);
+      String key = (state == null) ? SUMMARY_STATE_UNSCHEDULED : state.name();
+      byState.merge(key, 1, Integer::sum);
+      if (state == TaskPartitionState.COMPLETED) {
+        completed++;
+      } else if (isFailedState(state)) {
+        failed++;
+        failedTasks.add(p);
+        if (state == TaskPartitionState.TIMED_OUT) {
+          timedOut++;
+          timedOutTasks.add(p);
+        }
+      } else if (isInProgressState(state)) {
+        inProgress++;
+        inProgressTasks.add(p);
+      } else if (isPendingState(state)) {
+        pending++;
+        pendingTasks.add(p);
+      }
+    }
+    failedTasks.sort(null);
+    timedOutTasks.sort(null);
+    inProgressTasks.sort(null);
+    pendingTasks.sort(null);
+
+    Map<String, Object> summary = new LinkedHashMap<>();
+    summary.put(SUMMARY_KEY_TOTAL, partitions.size());
+    summary.put(SUMMARY_KEY_COMPLETED, completed);
+    summary.put(SUMMARY_KEY_FAILED, failed);
+    summary.put(SUMMARY_KEY_TIMED_OUT, timedOut);
+    summary.put(SUMMARY_KEY_IN_PROGRESS, inProgress);
+    summary.put(SUMMARY_KEY_PENDING, pending);
+    summary.put(SUMMARY_KEY_OTHER, partitions.size() - completed - failed - inProgress - pending);
+    summary.put(SUMMARY_KEY_BY_STATE, byState);
+    summary.put(SUMMARY_KEY_FAILED_TASKS, failedTasks);
+    summary.put(SUMMARY_KEY_TIMED_OUT_TASKS, timedOutTasks);
+    summary.put(SUMMARY_KEY_IN_PROGRESS_TASKS, inProgressTasks);
+    summary.put(SUMMARY_KEY_PENDING_TASKS, pendingTasks);
+
+    try {
+      String json = OBJECT_MAPPER.writeValueAsString(summary);
+      if (!json.equals(getTaskStatusSummary())) {
+        _record.setSimpleField(ContextProperties.TASK_STATUS_SUMMARY.name(), json);
+        markJobContextAsModified();
+      }
+    } catch (JsonProcessingException e) {
+      // The summary is a best-effort convenience field; never let it disrupt the job status
+      // update path.
+      LOG.warn("Failed to serialize task status summary for job {}", getName(), e);
+    }
+  }
+
+  /**
+   * @return the aggregated per-task status summary as a JSON string, or null if it has not been
+   *         computed yet (i.e. the job has not reached a terminal state). See
+   *         {@link #updateTaskStatusSummary()} for the shape.
+   */
+  public String getTaskStatusSummary() {
+    return _record.getSimpleField(ContextProperties.TASK_STATUS_SUMMARY.name());
+  }
+
   /**
    * Get MapField for the given partition.
    * @param p

helix-core/src/test/java/org/apache/helix/integration/task/TestJobTaskStatusSummary.java

@@ -0,0 +1,144 @@
+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.ArrayList;
+import java.util.List;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+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;
+
+/**
+ * Verifies the aggregated per-task status summary that Helix writes into the JobContext when a job
+ * reaches a terminal state. The scenario mirrors a common validation setup: one job per table, one
+ * task per partition, and a high FailureThreshold so every task runs even if some fail. In that
+ * setup the job's own status flag is COMPLETED, which previously masked partition level failures.
+ * The summary must surface those failures.
+ */
+public class TestJobTaskStatusSummary extends TaskTestBase {
+
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  @Test
+  public void testSummaryReflectsPartialFailureOnCompletedJob() throws Exception {
+    int numTasks = 6;
+    // Indices of the tasks that should fail terminally.
+    final int fatalTask = 1; // FATAL_FAILED -> given up immediately -> TASK_ABORTED
+    final int exceptionTask = 3; // throws -> retried to exhaustion -> TASK_ERROR
+    final int expectedFailed = 2;
+    final int expectedCompleted = numTasks - expectedFailed;
+
+    String jobResource = TestHelper.getTestMethodName();
+    JobConfig.Builder jobBuilder = new JobConfig.Builder();
+    // FailureThreshold high (a common operator workaround) so all tasks run and the job still
+    // ends COMPLETED.
+    jobBuilder.setCommand(MockTask.TASK_COMMAND).setTimeoutPerTask(10000).setMaxAttemptsPerTask(2)
+        .setFailureThreshold(Integer.MAX_VALUE);
+
+    List<TaskConfig> taskConfigs = new ArrayList<>();
+    for (int j = 0; j < numTasks; j++) {
+      TaskConfig.Builder configBuilder = new TaskConfig.Builder().setTaskId("task_" + j);
+      if (j == fatalTask) {
+        configBuilder.addConfig(MockTask.TASK_RESULT_STATUS, TaskResult.Status.FATAL_FAILED.name());
+      } else if (j == exceptionTask) {
+        configBuilder.addConfig(MockTask.THROW_EXCEPTION, Boolean.TRUE.toString());
+      }
+      configBuilder.setTargetPartition(String.valueOf(j));
+      taskConfigs.add(configBuilder.build());
+    }
+    jobBuilder.addTaskConfigs(taskConfigs);
+
+    Workflow flow =
+        WorkflowGenerator.generateSingleJobWorkflowBuilder(jobResource, jobBuilder).build();
+    _driver.start(flow);
+
+    // The job completes even though tasks failed, because FailureThreshold is high.
+    _driver.pollForWorkflowState(jobResource, TaskState.COMPLETED);
+
+    String namespacedJob = TaskUtil.getNamespacedJobName(jobResource);
+    Assert.assertEquals(_driver.getWorkflowContext(jobResource).getJobState(namespacedJob),
+        TaskState.COMPLETED, "Job status flag should be COMPLETED (this is what masked failures).");
+
+    JobContext ctx = _driver.getJobContext(namespacedJob);
+
+    // Cross-check the ground truth directly from per-partition states.
+    int actualCompleted = 0;
+    int actualFailed = 0;
+    for (int pId : ctx.getPartitionSet()) {
+      TaskPartitionState state = ctx.getPartitionState(pId);
+      if (state == TaskPartitionState.COMPLETED) {
+        actualCompleted++;
+      } else if (state == TaskPartitionState.TASK_ABORTED || state == TaskPartitionState.TASK_ERROR
+          || state == TaskPartitionState.TIMED_OUT || state == TaskPartitionState.ERROR) {
+        actualFailed++;
+      }
+    }
+    Assert.assertEquals(actualCompleted, expectedCompleted);
+    Assert.assertEquals(actualFailed, expectedFailed);
+
+    // Now the crux: the summary must be present and must match the ground truth.
+    String summaryJson = ctx.getTaskStatusSummary();
+    Assert.assertNotNull(summaryJson, "Task status summary should be populated on a terminal job.");
+    JsonNode summary = OBJECT_MAPPER.readTree(summaryJson);
+
+    Assert.assertEquals(summary.get("total").asInt(), numTasks);
+    Assert.assertEquals(summary.get("completed").asInt(), expectedCompleted);
+    Assert.assertEquals(summary.get("failed").asInt(), expectedFailed);
+    Assert.assertEquals(summary.get("other").asInt(), 0);
+    Assert.assertEquals(summary.get("failedTasks").size(), expectedFailed);
+
+    // No task timed out or stayed in-flight in this scenario, but the counts must be present so
+    // operators can rely on them being part of the summary shape.
+    Assert.assertEquals(summary.get("timedOut").asInt(), 0);
+    Assert.assertEquals(summary.get("inProgress").asInt(), 0);
+    Assert.assertEquals(summary.get("pending").asInt(), 0);
+    Assert.assertEquals(summary.get("timedOutTasks").size(), 0);
+    Assert.assertEquals(summary.get("inProgressTasks").size(), 0);
+    Assert.assertEquals(summary.get("pendingTasks").size(), 0);
+    // The top-level counts partition the tasks:
+    // total = completed + failed + inProgress + pending + other.
+    Assert.assertEquals(summary.get("completed").asInt() + summary.get("failed").asInt()
+        + summary.get("inProgress").asInt() + summary.get("pending").asInt()
+        + summary.get("other").asInt(), numTasks);
+
+    JsonNode byState = summary.get("byState");
+    Assert.assertEquals(byState.get(TaskPartitionState.COMPLETED.name()).asInt(), expectedCompleted);
+    Assert.assertEquals(byState.get(TaskPartitionState.TASK_ABORTED.name()).asInt(), 1);
+    Assert.assertEquals(byState.get(TaskPartitionState.TASK_ERROR.name()).asInt(), 1);
+
+    // The failed partition ids reported in the summary must actually be failed partitions.
+    for (JsonNode failedPidNode : summary.get("failedTasks")) {
+      TaskPartitionState state = ctx.getPartitionState(failedPidNode.asInt());
+      Assert.assertTrue(state != TaskPartitionState.COMPLETED,
+          "Partition " + failedPidNode.asInt() + " reported as failed but state is " + state);
+    }
+  }
+}