HttpRemoteTaskRunner can assign cancelled PENDING and PENDING_WORKER_ASSIGN tasks

Description

HttpRemoteTaskRunner.shutdown(taskId, reason) does not handle either PENDING or PENDING_WORKER_ASSIGN. This is a per-task cancellation call, not shutdown of the runner itself: the assignment loop remains active and can launch the cancelled task.

There are two cases worth tracking separately:

1. Cancellation while strictly PENDING: confirmed by CI logs. A task remains eligible for selection even after its status is persisted as FAILED and HRTR receives its shutdown request.
2. Cancellation while PENDING_WORKER_ASSIGN: confirmed missing handling in source; the specific concurrent interleaving has not been reproduced independently. Worker selection has already happened, and an assignment request may be about to start or already be in flight. A fix that only removes a pending ID cannot stop the assignment thread's captured work item.

Source and CI examined

• CI run: https://github.com/FrankChen021/druid/actions/runs/34215796434
• PR whose CI exposed the behavior: https://github.com/FrankChen021/druid/pull/196 (test workload changes, not an HRTR fix)
• CI head: ef8090de9586daffc08c851528c092982e782384
• Matching artifact: unit-test-reports-jdk25-55e4156d, artifact ID 10052676864: https://github.com/FrankChen021/druid/actions/runs/34215796434/artifacts/10052676864
• Report: TEST-org.apache.druid.testing.embedded.indexing.autoscaler.CostBasedAutoScalerIntegrationTest.xml
• Local source inspected: 33f7f778dfc2d8b06ca7c1d7a93f54f162526cee. Its HRTR and TaskQueue files have no diff against the CI head.

The affected source is verified at these commits; an earliest affected release has not been established.

Case 1: cancelled PENDING tasks are later assigned

Surefire reports 98.596 seconds for CostBasedAutoScalerIntegrationTest#test_autoScaler_computesOptimalTaskCountAndProducesScaleUp.

The preceding test left tasks for datasource_alifbaan. One exact task ID is index_kafka_datasource_alifbaan_7fd4f836c343694_ibbcnlhl. The artifact contains:

10:54:07.260 HRTR: Adding pending task[index_kafka_datasource_alifbaan_7fd4f836c343694_ibbcnlhl].
10:54:09.200 MetadataTaskStorage: Updating status ... status=FAILED ... errorMsg=Killing task for graceful shutdown
10:54:09.201 HRTR: Shutdown [index_kafka_datasource_alifbaan_7fd4f836c343694_ibbcnlhl] because: [Killing task for graceful shutdown]
10:54:09.354 HRTR: Assigning task [index_kafka_datasource_alifbaan_7fd4f836c343694_ibbcnlhl] to worker [10.1.1.207:8091]
10:54:09.359 HRTR: Task[index_kafka_datasource_alifbaan_7fd4f836c343694_ibbcnlhl] started RUNNING on worker[10.1.1.207:8091].

Timestamps above are from September 8, 2026; excerpts omit log prefixes and abbreviate the metadata line. This proves the shutdown call reached HRTR before assignment, rather than merely waiting in TaskQueue's asynchronous executor.

The subsequent sequence shows successive waves of cancelled work occupying capacity:

The new task's pending interval is 93.780 seconds, about 95.1% of the reported test duration. This strongly supports stale-task scheduling as the dominant delay. The exact runtime improvement still needs a fixed-code comparison; this report does not claim all test overhead would disappear.

Why reconciliation does not prevent this

TaskQueue.notifyStatus marks the queue entry complete, persists terminal status, and asynchronously calls runner shutdown.

HRTR.shutdown only handles RUNNING and COMPLETE. The pending item and ID remain available to the assignment loop.

When a worker reports RUNNING, taskAddedOrUpdated checks TaskStorage only if there is no in-memory item. The retained pending item therefore transitions to RUNNING without consulting the persisted failure. Later reconciliation can kill it now that it is RUNNING, but another pending cancelled task can take its slot.

Case 2: PENDING_WORKER_ASSIGN cancellation race

The assignment loop reserves a worker and sets PENDING_WORKER_ASSIGN under statusLock, then releases the lock before calling runTaskOnWorker() / WorkerHolder.assignTask().

A possible interleaving is:

1. Assignment thread selects the task and releases statusLock.
2. TaskQueue persists cancellation; HRTR receives shutdown while the item is PENDING_WORKER_ASSIGN and takes no action.
3. The assignment thread sends the request, or an already-sent request completes.
4. A worker RUNNING announcement is accepted through the retained item.

This follows from the source, but the CI excerpts above do not prove cancellation occurred in this intermediate state. This case needs a latch-controlled regression test.

Proposed staged fix and acceptance criteria

First, narrowly fix exact State.PENDING: under statusLock, detach the item from tasks, remove its ID from pendingTaskIds, and notify waiters. Outside the lock, complete the captured item's existing future through taskComplete(item, null, TaskStatus.failure(taskId, reason)).

Selection uses the same lock: if cancellation wins while strictly PENDING, the task cannot subsequently be selected. If selection wins, the item is PENDING_WORKER_ASSIGN and belongs to the separate fix. Use exact state equality, not isPending(), for this staged change.

Do not complete futures under statusLock: callbacks may run directly, and taskComplete explicitly prohibits holding the lock (see #6201). Do not pre-set COMPLETE and then invoke the current setResult, which attempts the same strict state transition again. Detaching first allows reuse of the existing completion path without that double transition. Preserve running-task completion callbacks and metrics.

Separately handle in-flight assignment: remember cancellation across dispatch and late announcements; check before dispatch; shut down an accepted task if cancellation races the request; prevent reverting a cancelled item to PENDING. A shutdown request sent before the worker receives assignment can miss the task. Worker reservation accounting must remain valid until acknowledgement or timeout, and duplicate completion/callback delivery must be prevented. This second design is not yet implemented or validated.

Suggested tests:

• Cancel strictly PENDING with no capacity, then add capacity and a live task. Assert the cancelled task is never assigned and the live task progresses.
• Assert the cancelled item's original future completes with FAILED and the reason, its pending/known entries are removed, and repeated shutdown emits no duplicate terminal notification.
• Verify callbacks execute without statusLock held; simple same-thread reentry alone is insufficient because Java monitors are reentrant.
• Pause after selection and during assignment to exercise PENDING_WORKER_ASSIGN cancellation, late RUNNING, assignment rejection, and duplicate terminal announcements.
• Verify worker reservation cleanup with multiple assignment threads.
• Re-run the autoscaler class including preceding-test teardown; an isolated scale-up method does not reliably reproduce the cross-test backlog.

Impact and related history

This uses production HRTR code. Plausible production consequences are wasted worker slots/resources, delayed valid tasks, and increased ingestion lag. CI directly demonstrates delayed scheduling; data corruption or duplicate publication has not been demonstrated.

#13558 changed shutdown from unconditional removal to retaining tasks except when COMPLETE, to preserve shutdown callbacks/metrics. It also introduced the existing HttpRemoteTaskRunnerTest.testShutdown assertion that a pending task remains known. That assertion does not verify cancellation or prevent later assignment. A wholesale revert risks restoring the original running-task callback issue.

No implementation changes or new test runs were performed for this report. Evidence consists of source inspection, historical patch inspection, and the downloaded CI artifact.