[api] Fix race condition causing queued tasks to execute after SequentialBatchIterator close

[juntaozhang]

Purpose

ThreadPoolUtils.SequentialBatchIterator.close() is supposed to cancel queued tasks and wait for running tasks. Under high load, however, a queued task can start executing before it is marked CANCELLED, which makes ThreadPoolUtilsTest#testCloseCancelsQueuedTasksAndWaitsUninterruptibly fail intermittently with
AtomicInteger(3) expected 2.
see https://github.com/apache/paimon/actions/runs/32989842446/job/98244565156?pr=9389

• Normal flow

  worker:  run task 0  ->  FINISHED
  worker:  run task 1  ->  blocks on allowSecondToExit
  closer:  close() starts
  closer:  cancel(task 0)  ->  no-op
  closer:  cancel(task 1)  ->  interrupt worker
  closer:  cancel(task 2)  ->  CREATED -> CANCELLED
  worker:  task 1 returns
  worker:  run task 2  ->  sees CANCELLED -> skips

Result: executions == 2, test passes.

• Abnormal flow

  worker:  run task 0  ->  FINISHED
  worker:  run task 1  ->  blocks on allowSecondToExit
  closer:  close() starts
  closer:  cancel(task 0)  ->  no-op
  closer:  cancel(task 1)  ->  interrupt worker
          <-- closer thread is preempted -->
  worker:  task 1 wakes and returns
  worker:  run task 2  ->  sees CREATED -> RUNNING -> executes processor
          <-- closer thread resumes -->
  closer:  cancel(task 2)  ->  sees RUNNING -> calls runner.interrupt(), but cannot stop a RUNNING task; must wait for it to finish

The root cause is that task 2 started executing before close() could mark it as CANCELLED, it runs to completion and executions becomes 3.

Tests

[JingsongLi]

+1