This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new ee7f50ac767 Retry task deletion when Helix fails to remove the job
context (#20009)
ee7f50ac767 is described below
commit ee7f50ac7675b52bd1ea191728d9a94e8b33472b
Author: Xiang Fu <[email protected]>
AuthorDate: Wed Oct 28 14:49:28 2026 -0700
Retry task deletion when Helix fails to remove the job context (#20009)
* 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.
* Release the task manager lock while a forced delete backs off
Each Helix deleteJob attempt still runs under the manager's lock, but the
backoff between attempts no longer holds it, so other callers are not
blocked during a retry. Also use static Mockito imports in the new tests.
* Retry non-forced task deletes the same way
A non-forced Helix deleteJob ends with the same job-context removal as a
forced one and can hit the same NotEmpty race, and its steps are also
idempotent. Retry every HelixException from deleteTask instead of only
forced deletes. Precondition failures such as deleting from a running
queue are IllegalStateException and still fail immediately.
* Raise the task delete retry backoff to 1 second
---
.../core/minion/PinotHelixTaskResourceManager.java | 36 ++++++++++++++-
.../minion/PinotHelixTaskResourceManagerTest.java | 54 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 2 deletions(-)
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..f6a4295b8ce 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 DELETE_JOB_MAX_ATTEMPTS = 3;
+ private static final long DELETE_JOB_BACKOFF_MS = 1000L;
+
private final TaskDriver _taskDriver;
private final PinotHelixResourceManager _helixResourceManager;
@@ -319,10 +324,16 @@ public class PinotHelixTaskResourceManager {
/// Delete a single task from the task queue. The task queue should be
/// stopped before deleting the task, otherwise it fails with exception.
///
+ /// A delete that fails with [HelixException] is retried a few times with a
growing backoff. 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 delete skips ZNodes that are already gone, so a retry
resumes where the failed attempt stopped.
+ /// Each attempt holds this manager's lock, but the backoff between attempts
does not.
+ ///
/// @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.
- public synchronized void deleteTask(String taskName, boolean forceDelete) {
+ public void deleteTask(String taskName, boolean forceDelete) {
String taskType = getTaskType(taskName);
String helixJobQueueName = getHelixJobQueueName(taskType);
if (forceDelete) {
@@ -330,7 +341,28 @@ public class PinotHelixTaskResourceManager {
} else {
LOGGER.info("Deleting task: {} from queue: {} of task type: {}",
taskName, helixJobQueueName, taskType);
}
- _taskDriver.deleteJob(helixJobQueueName, taskName, forceDelete);
+ for (int attempt = 1; ; attempt++) {
+ try {
+ synchronized (this) {
+ _taskDriver.deleteJob(helixJobQueueName, taskName, forceDelete);
+ }
+ return;
+ } catch (HelixException e) {
+ if (attempt >= DELETE_JOB_MAX_ATTEMPTS) {
+ throw e;
+ }
+ long backoffMs = DELETE_JOB_BACKOFF_MS * attempt;
+ LOGGER.warn("Attempt {} to 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;
+ }
+ }
+ }
}
/// 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..3018ac5f365 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;
@@ -49,6 +50,7 @@ import org.apache.pinot.spi.config.table.TenantConfig;
import org.apache.pinot.spi.utils.JsonUtils;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.apache.pinot.spi.utils.CommonConstants.TABLE_NAME;
@@ -58,7 +60,9 @@ import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;
@@ -1278,6 +1282,56 @@ public class PinotHelixTaskResourceManagerTest {
assertTrue(allTasks.containsKey(taskName1));
}
+ @DataProvider
+ public Object[][] forceDelete() {
+ return new Object[][]{{true}, {false}};
+ }
+
+ @Test(dataProvider = "forceDelete")
+ public void testDeleteTaskRetriesWhenHelixFailsToRemoveJob(boolean
forceDelete) {
+ 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
+ doThrow(new HelixException("Failed to delete job:
TaskQueue_TestTask_Task_TestTask_a"))
+ .doNothing()
+ .when(taskDriver).deleteJob("TaskQueue_TestTask", "Task_TestTask_a",
forceDelete);
+
+ mgr.deleteTask("Task_TestTask_a", forceDelete);
+
+ verify(taskDriver, times(2)).deleteJob("TaskQueue_TestTask",
"Task_TestTask_a", forceDelete);
+ }
+
+ @Test(dataProvider = "forceDelete")
+ public void testDeleteTaskThrowsAfterMaxAttempts(boolean forceDelete) {
+ 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");
+ doThrow(failure).when(taskDriver).deleteJob("TaskQueue_TestTask",
"Task_TestTask_a", forceDelete);
+
+ HelixException thrown = expectThrows(HelixException.class, () ->
mgr.deleteTask("Task_TestTask_a", forceDelete));
+ assertSame(thrown, failure);
+ verify(taskDriver,
times(PinotHelixTaskResourceManager.DELETE_JOB_MAX_ATTEMPTS))
+ .deleteJob("TaskQueue_TestTask", "Task_TestTask_a", forceDelete);
+ }
+
+ @Test
+ public void testDeleteTaskDoesNotRetryNonHelixFailure() {
+ TaskDriver taskDriver = mock(TaskDriver.class);
+ PinotHelixTaskResourceManager mgr =
+ new
PinotHelixTaskResourceManager(mock(PinotHelixResourceManager.class),
taskDriver);
+
+ // Helix rejects a non-forced delete from a running queue with
IllegalStateException; retrying cannot help
+ doThrow(new IllegalStateException("Queue TaskQueue_TestTask is still
running!"))
+ .when(taskDriver).deleteJob("TaskQueue_TestTask", "Task_TestTask_a",
false);
+
+ expectThrows(IllegalStateException.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]
