Avoid losing allocation futures

[trafalgarw]

Record allocation futures for retried tasks in PendingAllocationManager to ensure proper tracking and cancellation.

Description

...

Checklist

• You have the right to contribute the code/documentation/assets to this project, and you agree to contribute it under the terms of the project's license(s).
• All changes have been made on a separate branch (not master) in a fork.
• The whole project compiles correctly and all tests pass, i.e. gradlew clean build succeeds with the changes made, and without unavoidable compiler/toolchain warnings.
• New tests covering the change have been added, if it is possible / makes sense.
• The documentation has been extended, if necessary.
• The PR contains a proposal for a well-formed commit message that mentions all authors who contributed to the PR.
  (The commits in the PR will be squashed into one commit on the base branch, and your proposed message is intended to be used for this commit. See below for a template you can use for the message. See this blog post and this one for more on well-formed commit messages.)

Proposed squash commit message

Avoid losing allocation futures

Record allocation futures for retried tasks in `PendingAllocationManager` to ensure proper tracking and cancellation.

Co-authored-by: trafalgarw <mrfengyong@outlook.com>

[swltr]

@trafalgarw It's unclear what this is supposed to achieve. Does it fix an issue with the current code? Does it add a feature?

[trafalgarw]

The submitted task is not recorded. if created a new plant model, it will still run in kernel executor. Its client (vehicle controller) cannot be recycled by JVM. So, record allocation futures for retried tasks in PendingAllocationManager to ensure proper tracking and cancellation.

  /**
   * Moves all waiting allocations back into the incoming queue so they can be rechecked.
   */
  private void scheduleRetryWaitingAllocations() {
    for (AllocatorCommand.Allocate allocate : deferredAllocations) {
      kernelExecutor.submit(
          new AllocatorTask(
              reservationPool,
              deferredAllocations,
              allocationAdvisor,
              kernelExecutor,
              globalSyncObject,
              allocate
          )
      );
    }
    deferredAllocations.clear();
  }