[AMORO-4235] Recognize stale optimizer responses on the AMS side

[j1wonpark]

Why are the changes needed?

Tables intermittently get stuck during self-optimizing, and the AMS log is flooded with:

TaskRuntimeException: Task has been reset or not yet scheduled

The root cause is that the AMS can reset a task an optimizer is still working on, and then
reject that optimizer's valid response for it.

A task is reset (token cleared, status -> PLANNED) in two places:

• OptimizerKeeper, when a SCHEDULED task's ack does not arrive within
  optimizer.task-ack-timeout — even though the optimizer is still alive (the
  SCHEDULED + ackTimeout branch of buildSuspendingPredication does not check whether the
  owning token is still active).
• resetStaleTasksForThread, when the same optimizer thread polls again while one of its tasks
  is still ACKED.

After the reset, the optimizer's in-flight ack/complete arrives and TaskRuntime.validThread
sees token == null, so it throws. This:

• surfaces as an ERROR (PersistentBase "failed to commit transaction" plus the thrift layer), and
• for a completion that lands on a meanwhile-rescheduled task, breaks the SCHEDULED -> SUCCESS
  transition with IllegalTaskStateException.

In other words, a perfectly valid optimizer response is dropped with a noisy error.

The issue also reports the table becoming permanently stuck and uncancelable. That symptom could
not be reproduced from the excerpt logs and may need the full log to confirm, so this PR uses
"Relates to" rather than "Close".

Note: #4239 lowers the client-side log level for the same exception. This PR addresses the
server-side root cause instead, and additionally covers the completion path that #4239 does not.

Brief change log

TaskRuntime now recognizes a stale response by (token, threadId, status) instead of throwing
unconditionally in validThread:

• ack: if the task was reset/rescheduled, reject it outside the transaction so the
  optimizer skips the obsolete round, without the misleading "failed to commit transaction" error.
  The exception message is preserved so existing clients still recognize it.
• complete: if stale, ignore it gracefully — the reported run belongs to a torn-down round and
  the task will be re-executed in its current round. This also removes the
  IllegalTaskStateException variant and the equivalent race on canceled tasks.

A WARN line is logged in both cases (with status and owner) so the situation stays observable.

• attempt id guard (follow-up to review): (token, threadId, status) alone cannot tell two
  attempts of the same task on the same optimizer thread apart — once the rescheduled round is
  ACKED again, a delayed completion of the previous attempt matches all three. Every schedule now
  stamps a monotonically increasing task-attempt-id into the task properties; the optimizer
  echoes it back in the result summary and complete() ignores a completion reporting a different
  attempt. No thrift schema change: both OptimizingTask.properties and
  OptimizingTaskResult.summary are existing optional maps. Older optimizers do not echo the id
  and fall back to the previous checks (the same acceptance window as before this PR). The id is
  persisted with the task, so it stays monotonic across AMS restarts. Note: the attempt id shows
  up in the task properties on the dashboard task detail page.

How was this patch tested?

• Unit reproductions in TestOptimizingQueue via the real pollTask path: multi-task
  (stale ack is rejected) and single-task (stale completion is ignored).
• End-to-end reproduction in TestDefaultOptimizingService driving the real OptimizerKeeper:
  a live optimizer (kept alive by heartbeats) leaves a SCHEDULED task unacked past the ack
  timeout, the keeper resets it, and the late ack is rejected as expected. With the fix reverted,
  this test reproduces the issue's exact log sequence and stack trace -- same messages and line
  numbers, including the misleading "optimizer is expired" log for an optimizer that is in fact
  still alive.
• Full TestOptimizingQueue regression passes.
• TestOptimizingQueue#testStaleCompleteFromPreviousAttemptIsIgnored reproduces the
  reviewer-reported sequence (reset -> re-schedule to the same thread -> re-ack -> delayed
  completion of the first attempt): the stale completion is ignored and the current attempt's own
  completion is accepted. TestOptimizerExecutor covers the attempt-id echo for both success and
  error results.

Documentation

• Does this pull request introduce a new feature? No.

[czy006]

TaskRuntime.complete only ignores a completion when isStaleResponse(thread) || status != ACKED. If the old round’s completeTask RPC is delayed until the same task has already gone through SCHEDULED -> ACKED again on the same optimizer thread, then token, threadId, and status all match. The old completion will pass the current check and be applied as the result of the new round.

[j1wonpark]

Thanks for the careful review — that's a sharp observation, and you're right. When the same task is re-scheduled to the same optimizer thread, schedule() re-assigns the identical token/threadId, so (token, threadId, status) has no way to tell a delayed completion of the previous round apart from the current one.

For context on how I ended up leaving this window as-is in this PR: the behavior is the same before and after this change. The removed validThread relied on the same (token, threadId) key, and the state machine already accepted a thread-matching completion only at ACKED — so the pre-existing code accepted the same delayed completion in exactly the same window. This PR changes how the surrounding cases fail (graceful reject/ignore instead of a transaction error), but does not widen what gets accepted.

Closing the window completely would need a per-attempt discriminator that the optimizer echoes back (e.g. an attempt id carried via the existing OptimizingTask.properties / OptimizingTaskResult.summary maps, compared in complete()), which requires changes on the optimizer client side as well. Since this PR doesn't change what gets accepted, I'd prefer to keep that improvement out of its scope and keep this PR focused on handling the stale responses gracefully on the AMS side.

[czy006]

I think this still needs an attempt/generation guard for completions.

TaskRuntime.complete() currently treats a completion as current as long as (token, threadId) match and the task is in ACKED. That covers the new test where the stale completion arrives while the rescheduled task is still SCHEDULED, but it does not cover this sequence:

1. Task T is scheduled to optimizer token A, thread 1.
2. Optimizer acks T and starts the first attempt.
3. AMS resets T back to PLANNED.
4. The same optimizer thread polls T again, so T is scheduled again to token A, thread 1.
5. Optimizer acks the second attempt, so T is now ACKED again.
6. A delayed completion from the first attempt reaches AMS.

At step 6, the current checks cannot distinguish the first attempt from the second attempt, because the result only carries taskId/threadId and the live task state also has the same token/threadId with status ACKED. The stale completion can therefore be accepted as the result of the new attempt.

The normal optimizer loop is mostly serial, so this may be rare, but the server-side protocol is still not generation-safe under delayed RPCs, HA/client timeout/retry, or shutdown drain edges.

A safer fix would be to add an attempt id / schedule generation when the task is scheduled, include it in OptimizingTask, and require completeTask to return the same generation before accepting the result. Alternatively, the code needs a stronger guarantee that a reset task cannot be rescheduled to the same optimizer thread while a previous completion may still arrive.

cc WDYT? @j1wonpark

[j1wonpark]

Good point, and agreed — implemented in f46b07c, following your option (a).

TaskRuntime.schedule() now stamps a monotonically increasing attempt id (task-attempt-id) into the task properties, which reach the optimizer through the existing OptimizingTask.properties map. The optimizer echoes it back in OptimizingTaskResult.summary, and complete() ignores a completion whose echoed attempt id does not match the current one — exactly the step-6 case you described. The echo sits at the single point every completion passes through (the poll loop in OptimizerExecutor), so the standalone/Flink/Spark optimizers are all covered, including error results built by executor subclasses.

No thrift schema change is needed — both maps are existing optional fields. Compatibility:

• An older optimizer doesn't echo the id, so its results fall back to the (token, threadId, status) checks: the same acceptance window as before this PR, no regression.
• The attempt id is persisted with the task (the existing properties column), so it stays monotonic across AMS restarts.

One remaining gap, for the record: the ackTask RPC cannot carry the attempt id without a thrift signature change, so acks are still matched by (token, threadId, status) only. Since an ack is sent once and not retried (unlike completions, which are drained with retries on shutdown), the completion path is where this ambiguity actually bites; I'd leave the ack side out of this PR.

Tests: TestOptimizingQueue#testStaleCompleteFromPreviousAttemptIsIgnored reproduces your exact sequence and asserts the stale completion is ignored while the current attempt's own completion is accepted; TestOptimizerExecutor#testEchoTaskAttemptIdOnSuccess/OnFailure cover the echo.

[czy006]

on the default 30-second ack timeout and then sleeps for 35 seconds. This adds a large fixed delay to the test suite and can still be brittle under slow or noisy CI environments.

Suggested fix: reduce optimizer.task-ack-timeout for this test to a small test-only value, such as 1-2 seconds, and wait for the task status with polling/await logic instead of a fixed 35-second sleep.

[j1wonpark]

Good point, fixed in 515be61. OPTIMIZER_TASK_ACK_TIMEOUT is now overridden in AMSServiceTestBase and the fixed 35s sleep is replaced with a 100ms status-polling wait, so the test finishes right after the keeper reset lands (~6s instead of 35s).

One finding from trying the suggested 1-2s value: the ack timeout has to stay above optimizer.polling-timeout (3s). A pollTask blocking for its full polling timeout would otherwise pick up the very task the keeper reset by ack-timeout in the meantime — at 2s this broke testPollTaskThreeTimes, whose final poll expects null. So I settled on 5s and left a comment in AMSServiceTestBase documenting the constraint.