xiangfu0 commented on code in PR #20009:
URL: https://github.com/apache/pinot/pull/20009#discussion_r4365183685
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java:
##########
@@ -322,15 +327,46 @@ public synchronized void deleteTaskQueue(String taskType,
boolean forceDelete) {
/// @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);
Review Comment:
Changed in 1b1d56e30e: `deleteTask` is no longer `synchronized` as a whole.
Each Helix `deleteJob` attempt runs inside `synchronized (this)`, and the
backoff sleeps outside the lock, so other callers of this manager are not
blocked while a forced delete waits to retry.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManagerTest.java:
##########
@@ -1278,6 +1279,50 @@ public void testGetTaskCountsWithTableFilterEdgeCases() {
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"))
Review Comment:
Done in 1b1d56e30e: the new tests use static `doThrow` and `times`.
`doNothing()` is chained on the stubber, so it needs no import.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
