This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/fix-force-delete-task-notempty-race in repository https://gitbox.apache.org/repos/asf/pinot.git
commit d91f078563efa9519979e503f8e082335c8b7bd7 Author: Xiang Fu <[email protected]> AuthorDate: Thu Oct 22 23:47:58 2026 -0700 Retry forced task deletion when Helix fails to remove the job context A forced Helix deleteJob can race with the task pipeline, which may still be writing the job context right after a task is scheduled or its queue is stopped. Removing the context ZNode then fails with ZooKeeper NotEmpty, Helix throws a cause-less HelixException, and table deletion returns 500. Retry a forced delete up to three times with a short backoff. Each step of Helix's forced delete skips ZNodes that are already gone, so a retry resumes where the failed attempt stopped. --- .../core/minion/PinotHelixTaskResourceManager.java | 38 +++++++++++++++++- .../minion/PinotHelixTaskResourceManagerTest.java | 45 ++++++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java index 5a3ab6fd40f..704ba16e1cb 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java @@ -42,6 +42,7 @@ import javax.annotation.Nullable; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.tuple.Pair; import org.apache.hc.client5.http.io.HttpClientConnectionManager; +import org.apache.helix.HelixException; import org.apache.helix.model.InstanceConfig; import org.apache.helix.store.zk.ZkHelixPropertyStore; import org.apache.helix.task.JobConfig; @@ -91,6 +92,10 @@ public class PinotHelixTaskResourceManager { private static final EnumSet<TaskState> TERMINAL_STATES = EnumSet.of(TaskState.COMPLETED, TaskState.FAILED, TaskState.TIMED_OUT, TaskState.ABORTED); + @VisibleForTesting + static final int FORCE_DELETE_JOB_MAX_ATTEMPTS = 3; + private static final long FORCE_DELETE_JOB_BACKOFF_MS = 100L; + private final TaskDriver _taskDriver; private final PinotHelixResourceManager _helixResourceManager; @@ -322,15 +327,46 @@ public class PinotHelixTaskResourceManager { /// @param taskName the task to delete from the queue. /// @param forceDelete as said in helix comment, if set true, all job's related zk nodes will /// be clean up from zookeeper even if its workflow information can not be found. + /// A forced delete that fails is retried a few times with a short backoff. public synchronized void deleteTask(String taskName, boolean forceDelete) { String taskType = getTaskType(taskName); String helixJobQueueName = getHelixJobQueueName(taskType); if (forceDelete) { LOGGER.warn("Force deleting task: {} from queue: {} of task type: {}", taskName, helixJobQueueName, taskType); + forceDeleteJob(helixJobQueueName, taskName); } else { LOGGER.info("Deleting task: {} from queue: {} of task type: {}", taskName, helixJobQueueName, taskType); + _taskDriver.deleteJob(helixJobQueueName, taskName, false); + } + } + + /// Force deletes a job, retrying when Helix fails to remove its ZNodes. + /// + /// The Helix task pipeline can still be writing the job context right after the queue is stopped or the task is + /// scheduled. Removing the context ZNode then fails with ZooKeeper `NotEmpty`, which Helix reports as a + /// [HelixException] without the cause. Each step of a forced delete skips ZNodes that are already gone, so a retry + /// resumes where the failed attempt stopped. The caller holds this manager's lock, so the backoff stays short. + private void forceDeleteJob(String helixJobQueueName, String taskName) { + for (int attempt = 1; ; attempt++) { + try { + _taskDriver.deleteJob(helixJobQueueName, taskName, true); + return; + } catch (HelixException e) { + if (attempt >= FORCE_DELETE_JOB_MAX_ATTEMPTS) { + throw e; + } + long backoffMs = FORCE_DELETE_JOB_BACKOFF_MS * attempt; + LOGGER.warn("Attempt {} to force delete task: {} from queue: {} failed, retrying in {}ms", attempt, taskName, + helixJobQueueName, backoffMs, e); + try { + Thread.sleep(backoffMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + e.addSuppressed(ie); + throw e; + } + } } - _taskDriver.deleteJob(helixJobQueueName, taskName, forceDelete); } /// Get all task queues. diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManagerTest.java index fbecb58482e..c01360364e0 100644 --- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManagerTest.java +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManagerTest.java @@ -29,6 +29,7 @@ import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; +import org.apache.helix.HelixException; import org.apache.helix.task.JobConfig; import org.apache.helix.task.JobContext; import org.apache.helix.task.JobDag; @@ -1278,6 +1279,50 @@ public class PinotHelixTaskResourceManagerTest { assertTrue(allTasks.containsKey(taskName1)); } + @Test + public void testForceDeleteTaskRetriesWhenHelixFailsToRemoveJob() { + TaskDriver taskDriver = mock(TaskDriver.class); + PinotHelixTaskResourceManager mgr = + new PinotHelixTaskResourceManager(mock(PinotHelixResourceManager.class), taskDriver); + + // Helix reports a ZooKeeper NotEmpty failure while removing the job context as a cause-less HelixException + Mockito.doThrow(new HelixException("Failed to delete job: TaskQueue_TestTask_Task_TestTask_a")) + .doNothing() + .when(taskDriver).deleteJob("TaskQueue_TestTask", "Task_TestTask_a", true); + + mgr.deleteTask("Task_TestTask_a", true); + + verify(taskDriver, Mockito.times(2)).deleteJob("TaskQueue_TestTask", "Task_TestTask_a", true); + } + + @Test + public void testForceDeleteTaskThrowsAfterMaxAttempts() { + TaskDriver taskDriver = mock(TaskDriver.class); + PinotHelixTaskResourceManager mgr = + new PinotHelixTaskResourceManager(mock(PinotHelixResourceManager.class), taskDriver); + + HelixException failure = new HelixException("Failed to delete job: TaskQueue_TestTask_Task_TestTask_a"); + Mockito.doThrow(failure).when(taskDriver).deleteJob("TaskQueue_TestTask", "Task_TestTask_a", true); + + HelixException thrown = expectThrows(HelixException.class, () -> mgr.deleteTask("Task_TestTask_a", true)); + assertSame(thrown, failure); + verify(taskDriver, Mockito.times(PinotHelixTaskResourceManager.FORCE_DELETE_JOB_MAX_ATTEMPTS)) + .deleteJob("TaskQueue_TestTask", "Task_TestTask_a", true); + } + + @Test + public void testDeleteTaskWithoutForceDoesNotRetry() { + TaskDriver taskDriver = mock(TaskDriver.class); + PinotHelixTaskResourceManager mgr = + new PinotHelixTaskResourceManager(mock(PinotHelixResourceManager.class), taskDriver); + + Mockito.doThrow(new HelixException("Workflow TaskQueue_TestTask is not stopped")) + .when(taskDriver).deleteJob("TaskQueue_TestTask", "Task_TestTask_a", false); + + expectThrows(HelixException.class, () -> mgr.deleteTask("Task_TestTask_a", false)); + verify(taskDriver).deleteJob("TaskQueue_TestTask", "Task_TestTask_a", false); + } + // ==================== Tests for task queue bounding ==================== @Test --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
