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:

```text
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:

| Time | Event |
|---|---|
| 10:54:13.073 | New test task `index_kafka_datasource_diinaocp_6d1c6a3fba1682a_mpplelhh` enters HRTR pending state. |
| 10:54:46.801 | Reconciliation invokes HRTR shutdown again for old tasks, including `index_kafka_datasource_alifbaan_c8be674ffa9e8b8_mdlpbdho`. |
| 10:54:46.812 | HRTR assigns that already-cancelled task. Two more old tasks are assigned at `.819` and `.827`; all three are reported RUNNING by `.835`. |
| 10:55:46.800 onward | Reconciliation requests shutdown again. |
| 10:55:46.848 | One more cancelled old task is assigned. |
| 10:55:46.853 | The new test task finally gets assigned. |
| 10:55:47.091 | Autoscaler selects task count `4`. |

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](https://github.com/FrankChen021/druid/blob/ef8090de9586daffc08c851528c092982e782384/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java#L710) marks the queue entry complete, persists terminal status, and asynchronously calls runner shutdown.

[HRTR.shutdown](https://github.com/FrankChen021/druid/blob/ef8090de9586daffc08c851528c092982e782384/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java#L1241) only handles `RUNNING` and `COMPLETE`. The pending item and ID remain available to the [assignment loop](https://github.com/FrankChen021/druid/blob/ef8090de9586daffc08c851528c092982e782384/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java#L1090).

When a worker reports `RUNNING`, [taskAddedOrUpdated](https://github.com/FrankChen021/druid/blob/ef8090de9586daffc08c851528c092982e782384/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java#L1431) 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.

Start with HttpRemoteTaskRunner.shutdown, the assignment loop, and taskAddedOrUpdated, then read the existing HttpRemoteTaskRunnerTest.testShutdown coverage. Reproduce the strictly PENDING cancellation case with a focused regression test and run the autoscaler integration test mentioned in the report. Treat PENDING_WORKER_ASSIGN cancellation as a separate design and test effort; completion requires preserving callbacks, future results, lock safety, and worker reservation cleanup.

Written by the indexing model from the issue text.