This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/helix.git
commit c928d0b0d9ad586611490af6ca2c48b54b26261f Author: Junkai Xue <[email protected]> AuthorDate: Tue Nov 10 21:43:02 2026 -0800 Take back unstarted tasks and handle unavailable instances while a job is FAILING (#3202) Follow-up to the change that stops a FAILING job from getting new tasks. The places that treated a TIMING_OUT job specially now use a shared check that also covers FAILING: - a task that is still pending on INIT->RUNNING is set back to INIT on every pipeline, not only on the first one, and a task left in INIT is dropped and its slot released; - a task in flight on an instance that is no longer among the live instances of the job (gone, disabled or no longer eligible) is dropped. The controller does not go through the tasks of such an instance, so the task is never aborted like the others. Assigning tasks used to handle part of this as a side effect; a job that is failing or timing out now does it itself: the instance is asked to drop the task while it still has a current state for it, and the task is marked DROPPED in the context once there is neither a current state nor a pending message. TIMING_OUT had the same gap. TestJobFailingNoNewTasks no longer depends on how fast the first batch of tasks starts, can run more than once in a JVM, and also checks that the job fails when the instance of a running task is disabled. The task factories of the test participants can be overridden through TaskSynchronizedTestBase.createTaskFactoryRegistry(). --- .../apache/helix/task/AbstractTaskDispatcher.java | 20 +++- .../java/org/apache/helix/task/JobDispatcher.java | 50 ++++++++- .../integration/task/TestJobFailingNoNewTasks.java | 113 ++++++++++++++++----- .../helix/task/TaskSynchronizedTestBase.java | 13 ++- 4 files changed, 158 insertions(+), 38 deletions(-) diff --git a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java index 5d3364803..ac9ebafb7 100644 --- a/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java +++ b/helix-core/src/main/java/org/apache/helix/task/AbstractTaskDispatcher.java @@ -291,10 +291,10 @@ public abstract class AbstractTaskDispatcher { // timing out. In this case, it will be sent back to INIT state to be removed. Here we // ensure that this task then goes from INIT to DROPPED so that it will be released from // AssignableInstance to prevent resource leak - if (jobState == TaskState.TIMED_OUT || jobState == TaskState.TIMING_OUT + if (jobState == TaskState.TIMED_OUT || isJobAbortingTasks(jobState) || jobTgtState == TargetState.DELETE) { - // Job is timed out or timing out or targetState is to be deleted, so its tasks will be - // sent back to INIT + // Job is timed out, timing out or failing, or targetState is to be deleted, so its + // tasks will be sent back to INIT // In this case, tasks' IdealState will be removed, and they will be sent to DROPPED partitionsToDropFromIs.add(pId); @@ -448,9 +448,10 @@ public abstract class AbstractTaskDispatcher { Message pendingMessage, TaskState jobState, TaskPartitionState currState, Map<Integer, PartitionAssignment> paMap, Map<String, Set<Integer>> assignedPartitions) { - if (jobState == TaskState.TIMING_OUT && currState == TaskPartitionState.INIT + if (isJobAbortingTasks(jobState) && currState == TaskPartitionState.INIT && pendingMessage.getToState().equals(TaskPartitionState.RUNNING.name())) { - // While job is timing out, if the task is pending on INIT->RUNNING, set it back to INIT, + // While job is timing out or failing, if the task is pending on INIT->RUNNING, set it back + // to INIT, // so that Helix will cancel the transition. paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.INIT.name())); assignedPartitions.get(instance).add(pId); @@ -846,6 +847,15 @@ public abstract class AbstractTaskDispatcher { return filteredTasks; } + /** + * Returns whether the job only waits for its tasks in flight to be aborted before it reaches its + * terminal state. Such a job must not get new tasks, and its tasks that have not started + * running yet are taken back. + */ + protected static boolean isJobAbortingTasks(TaskState jobState) { + return jobState == TaskState.TIMING_OUT || jobState == TaskState.FAILING; + } + /** * Returns whether if the task is not in a terminal state and could be re-scheduled. * @param state diff --git a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java index 93bd8fb00..e6d8f8162 100644 --- a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java +++ b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java @@ -157,7 +157,7 @@ public class JobDispatcher extends AbstractTaskDispatcher { || TaskState.TIMED_OUT.equals(workflowState))) { jobState = TaskState.TIMING_OUT; workflowCtx.setJobState(jobName, TaskState.TIMING_OUT); - } else if (jobState != TaskState.TIMING_OUT && jobState != TaskState.FAILING) { + } else if (!isJobAbortingTasks(jobState)) { // TIMING_OUT/FAILING/ABORTING job can't be stopped, because all tasks are being aborted // Update running status in workflow context if (jobTgtState == TargetState.STOP) { @@ -264,6 +264,9 @@ public class JobDispatcher extends AbstractTaskDispatcher { || (jobCfg.getTargetResource() != null && cache.getIdealState(jobCfg.getTargetResource()) != null && !cache.getIdealState(jobCfg.getTargetResource()).isEnabled())) { + Map<Integer, PartitionAssignment> dropAssignments = + dropTasksInFlightOnUnavailableInstances(jobResource, jobCtx, liveInstances, + currStateOutput); if (isJobFinished(jobCtx, jobResource, currStateOutput)) { failJob(jobResource, workflowCtx, jobCtx, workflowConfig, cache.getJobConfigMap(), cache); return buildEmptyAssignment(jobResource, currStateOutput); @@ -284,10 +287,19 @@ public class JobDispatcher extends AbstractTaskDispatcher { paMap.put(pId, new PartitionAssignment(instance, TaskPartitionState.INIT.name())); } } + // A task on an unavailable instance cannot be aborted, it is dropped instead. + paMap.putAll(dropAssignments); return toResourceAssignment(jobResource, paMap); } + if (isJobAbortingTasks(jobState)) { + // Do not replace an assignment that has already been made for a task, e.g. a drop that was + // requested by another instance. + dropTasksInFlightOnUnavailableInstances(jobResource, jobCtx, liveInstances, currStateOutput) + .forEach(paMap::putIfAbsent); + } + if (jobState == TaskState.FAILING && isJobFinished(jobCtx, jobResource, currStateOutput)) { failJob(jobResource, workflowCtx, jobCtx, workflowConfig, cache.getJobConfigMap(), cache); return buildEmptyAssignment(jobResource, currStateOutput); @@ -323,8 +335,8 @@ public class JobDispatcher extends AbstractTaskDispatcher { // Make additional task assignments if needed. A job that is timing out or failing only waits // for its tasks in flight to be aborted; a new task would be aborted on the next pipeline and // would keep the job from finishing until every task has been started once. - if (jobState != TaskState.TIMING_OUT && jobState != TaskState.TIMED_OUT - && jobState != TaskState.FAILING && jobTgtState == TargetState.START) { + if (!isJobAbortingTasks(jobState) && jobState != TaskState.TIMED_OUT + && jobTgtState == TargetState.START) { handleAdditionalTaskAssignment(currentInstanceToTaskAssignments, excludedInstances, jobResource, currStateOutput, jobCtx, jobCfg, workflowConfig, workflowCtx, cache, assignedPartitions, paMap, skippedPartitions, taskAssignmentCal, allPartitions, @@ -346,6 +358,38 @@ public class JobDispatcher extends AbstractTaskDispatcher { return ra; } + /** + * Takes care of the tasks in flight whose instance is not among the live instances of the job + * anymore, because it is gone, disabled or no longer eligible. The controller does not go + * through the tasks of such an instance, so they would never be aborted, and a job that is + * failing or timing out would wait for them forever. Normally this is handled when new tasks are + * assigned, which does not happen for such a job. + * If the instance still has a current state for the task, it is asked to drop it. Once there + * is neither a current state nor a pending message, the task is marked DROPPED in the context. + * @return the DROPPED assignments to make, by task partition + */ + private Map<Integer, PartitionAssignment> dropTasksInFlightOnUnavailableInstances( + String jobResource, JobContext jobCtx, Collection<String> liveInstances, + CurrentStateOutput currStateOutput) { + Map<Integer, PartitionAssignment> dropAssignments = new HashMap<>(); + for (int pId : jobCtx.getPartitionSet()) { + TaskPartitionState state = jobCtx.getPartitionState(pId); + String instance = jobCtx.getAssignedParticipant(pId); + if ((state != TaskPartitionState.RUNNING && state != TaskPartitionState.INIT) + || instance == null || liveInstances.contains(instance)) { + continue; + } + Partition partition = new Partition(pName(jobResource, pId)); + if (currStateOutput.getCurrentState(jobResource, partition, instance) != null) { + dropAssignments.put(pId, + new PartitionAssignment(instance, TaskPartitionState.DROPPED.name())); + } else if (currStateOutput.getPendingMessage(jobResource, partition, instance) == null) { + jobCtx.setPartitionState(pId, TaskPartitionState.DROPPED); + } + } + return dropAssignments; + } + private boolean isJobFinished(JobContext jobContext, String jobResource, CurrentStateOutput currentStateOutput) { for (int pId : jobContext.getPartitionSet()) { diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestJobFailingNoNewTasks.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestJobFailingNoNewTasks.java index 30899952e..5db6d06c5 100644 --- a/helix-core/src/test/java/org/apache/helix/integration/task/TestJobFailingNoNewTasks.java +++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestJobFailingNoNewTasks.java @@ -24,19 +24,18 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.apache.helix.TestHelper; -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.TaskConstants; import org.apache.helix.task.TaskFactory; +import org.apache.helix.task.TaskPartitionState; import org.apache.helix.task.TaskResult; import org.apache.helix.task.TaskState; -import org.apache.helix.task.TaskStateModelFactory; import org.apache.helix.task.TaskUtil; import org.apache.helix.task.Workflow; import org.testng.Assert; @@ -51,36 +50,28 @@ public class TestJobFailingNoNewTasks extends TaskTestBase { private static final int NUM_TASKS = 60; // One task per instance can run at the same time. private static final int NUM_SLOTS = 5; + // How long the tasks of the first batch ignore the cancellation after the first failure. + private static final long UNCANCELABLE_MS = 10000L; private static final AtomicInteger NUM_STARTED_TASKS = new AtomicInteger(0); + // When the first task failed, 0 if it has not failed yet. + private static final AtomicLong FIRST_FAILURE_TIME = new AtomicLong(0L); @Override - protected void startParticipant(String zkAddr, int i) { + protected Map<String, TaskFactory> createTaskFactoryRegistry() { Map<String, TaskFactory> taskFactoryReg = new HashMap<>(); taskFactoryReg.put(TASK_COMMAND, context -> new FailFirstTask()); - String instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + i); - _participants[i] = new MockParticipantManager(zkAddr, CLUSTER_NAME, instanceName); - StateMachineEngine stateMachine = _participants[i].getStateMachineEngine(); - stateMachine.registerStateModelFactory(TaskConstants.STATE_MODEL_NAME, - new TaskStateModelFactory(_participants[i], taskFactoryReg)); - _participants[i].syncStart(); + return taskFactoryReg; } @Test public void testNoNewTasksScheduledWhileJobIsFailing() throws Exception { Assert.assertEquals(_numNodes, NUM_SLOTS); + resetTaskCounters(); final String workflowName = TestHelper.getTestMethodName(); final String jobName = "testJob"; - // Many more tasks than the cluster can run at the same time. - List<TaskConfig> taskConfigs = new ArrayList<>(); - for (int i = 0; i < NUM_TASKS; i++) { - taskConfigs.add(new TaskConfig(TASK_COMMAND, new HashMap<>())); - } - JobConfig.Builder jobBuilder = - new JobConfig.Builder().setWorkflow(workflowName).setCommand(TASK_COMMAND) - .addTaskConfigs(taskConfigs).setNumConcurrentTasksPerInstance(1) - .setFailureThreshold(0); - _driver.start(new Workflow.Builder(workflowName).addJob(jobName, jobBuilder).build()); + _driver.start(new Workflow.Builder(workflowName).addJob(jobName, createJobBuilder(workflowName)) + .build()); String namespacedJobName = TaskUtil.getNamespacedJobName(workflowName, jobName); _driver.pollForJobState(workflowName, namespacedJobName, TaskState.FAILED); @@ -103,19 +94,75 @@ public class TestJobFailingNoNewTasks extends TaskTestBase { String.format("%d tasks were scheduled, expected at most %d", numScheduled, NUM_SLOTS)); } + /** + * The instance of a task that is in flight is disabled while the job is FAILING. The controller + * does not go through the tasks of a disabled instance anymore, so the task is never aborted + * like the others; it has to be dropped for the job to become FAILED. + */ + @Test(dependsOnMethods = "testNoNewTasksScheduledWhileJobIsFailing") + public void testFailingJobFinishesWhenInstanceOfRunningTaskIsDisabled() throws Exception { + resetTaskCounters(); + final String workflowName = TestHelper.getTestMethodName(); + final String jobName = "testJob"; + _driver.start(new Workflow.Builder(workflowName).addJob(jobName, createJobBuilder(workflowName)) + .build()); + + String namespacedJobName = TaskUtil.getNamespacedJobName(workflowName, jobName); + _driver.pollForJobState(workflowName, namespacedJobName, TaskState.FAILING); + + // The tasks of the first batch ignore the cancellation for a while after the first failure, + // so some of them are still running. The context may not show them as RUNNING right away. + AtomicReference<String> runningTaskInstance = new AtomicReference<>(); + Assert.assertTrue(TestHelper.verify(() -> { + JobContext jobContext = _driver.getJobContext(namespacedJobName); + for (int pId : jobContext.getPartitionSet()) { + if (jobContext.getPartitionState(pId) == TaskPartitionState.RUNNING) { + runningTaskInstance.set(jobContext.getAssignedParticipant(pId)); + return true; + } + } + return false; + }, UNCANCELABLE_MS / 2), "No task is running while the job is FAILING"); + String instanceOfRunningTask = runningTaskInstance.get(); + try { + _gSetupTool.getClusterManagementTool() + .enableInstance(CLUSTER_NAME, instanceOfRunningTask, false); + _driver.pollForJobState(workflowName, namespacedJobName, TaskState.FAILED); + _driver.pollForWorkflowState(workflowName, TaskState.FAILED); + } finally { + _gSetupTool.getClusterManagementTool() + .enableInstance(CLUSTER_NAME, instanceOfRunningTask, true); + } + } + + private static void resetTaskCounters() { + NUM_STARTED_TASKS.set(0); + FIRST_FAILURE_TIME.set(0L); + } + + private JobConfig.Builder createJobBuilder(String workflowName) { + // Many more tasks than the cluster can run at the same time. + List<TaskConfig> taskConfigs = new ArrayList<>(); + for (int i = 0; i < NUM_TASKS; i++) { + taskConfigs.add(new TaskConfig(TASK_COMMAND, new HashMap<>())); + } + return new JobConfig.Builder().setWorkflow(workflowName).setCommand(TASK_COMMAND) + .addTaskConfigs(taskConfigs).setNumConcurrentTasksPerInstance(1).setFailureThreshold(0); + } + /** * The behavior of a task depends on the order in which the tasks are started, so that the test * does not depend on which tasks the controller picks first: - * - the first task fails right away, which makes the job FAILING; - * - the other tasks of the first batch ignore the cancellation for a while, which keeps the job - * FAILING for several pipelines while the slot of the failed task is free; + * - the first task fails once all the slots are busy, which makes the job FAILING; + * - the other tasks of the first batch ignore the cancellation for a while after that failure, + * which keeps the job FAILING for several pipelines while the slot of the failed task is free; * - any later task (there is none unless the controller keeps scheduling) honors the * cancellation right away. */ private static class FailFirstTask implements Task { private static final long FAIL_DELAY_MS = 300L; - private static final long UNCANCELABLE_MS = 3000L; - private static final long MAX_RUN_MS = 30000L; + private static final long WAIT_FOR_FIRST_BATCH_MS = 20000L; + private static final long MAX_RUN_MS = 120000L; private volatile boolean _canceled = false; @Override @@ -123,13 +170,17 @@ public class TestJobFailingNoNewTasks extends TaskTestBase { int order = NUM_STARTED_TASKS.getAndIncrement(); try { if (order == 0) { + long deadline = System.currentTimeMillis() + WAIT_FOR_FIRST_BATCH_MS; + while (NUM_STARTED_TASKS.get() < NUM_SLOTS && System.currentTimeMillis() < deadline) { + Thread.sleep(10L); + } Thread.sleep(FAIL_DELAY_MS); + FIRST_FAILURE_TIME.set(System.currentTimeMillis()); return new TaskResult(TaskResult.Status.FATAL_FAILED, "The first task fails"); } long start = System.currentTimeMillis(); - long uncancelableUntil = start + (order < NUM_SLOTS ? UNCANCELABLE_MS : 0L); while (System.currentTimeMillis() < start + MAX_RUN_MS) { - if (_canceled && System.currentTimeMillis() >= uncancelableUntil) { + if (_canceled && (order >= NUM_SLOTS || isUncancelableWindowOver())) { return new TaskResult(TaskResult.Status.CANCELED, "Canceled"); } Thread.sleep(10L); @@ -141,6 +192,12 @@ public class TestJobFailingNoNewTasks extends TaskTestBase { return new TaskResult(TaskResult.Status.COMPLETED, "Completed"); } + private static boolean isUncancelableWindowOver() { + long firstFailureTime = FIRST_FAILURE_TIME.get(); + return firstFailureTime > 0 + && System.currentTimeMillis() >= firstFailureTime + UNCANCELABLE_MS; + } + @Override public void cancel() { _canceled = true; diff --git a/helix-core/src/test/java/org/apache/helix/task/TaskSynchronizedTestBase.java b/helix-core/src/test/java/org/apache/helix/task/TaskSynchronizedTestBase.java index 026ee3ccf..6879ce7c0 100644 --- a/helix-core/src/test/java/org/apache/helix/task/TaskSynchronizedTestBase.java +++ b/helix-core/src/test/java/org/apache/helix/task/TaskSynchronizedTestBase.java @@ -175,8 +175,7 @@ public class TaskSynchronizedTestBase extends ZkTestBase { stopParticipant(i); } - Map<String, TaskFactory> taskFactoryReg = new HashMap<>(); - taskFactoryReg.put(MockTask.TASK_COMMAND, MockTask::new); + Map<String, TaskFactory> taskFactoryReg = createTaskFactoryRegistry(); String instanceName = PARTICIPANT_PREFIX + "_" + (_startPort + i); _participants[i] = new MockParticipantManager(zkAddr, CLUSTER_NAME, instanceName); @@ -187,6 +186,16 @@ public class TaskSynchronizedTestBase extends ZkTestBase { _participants[i].syncStart(); } + /** + * The task factories registered on every participant, keyed by task command. Tests that need + * their own task implementation override this. + */ + protected Map<String, TaskFactory> createTaskFactoryRegistry() { + Map<String, TaskFactory> taskFactoryReg = new HashMap<>(); + taskFactoryReg.put(MockTask.TASK_COMMAND, MockTask::new); + return taskFactoryReg; + } + protected void stopParticipants() { for (int i = 0; i < _numNodes; i++) { stopParticipant(i);
