[SPARK-51436][CORE][SQL][K8s][SS] Fix bug that cancel Future specified mayInterruptIfRunning with true

[beliefer]

What changes were proposed in this pull request?

This PR proposes to fix bug that cancel Future specified mayInterruptIfRunning with true.

Why are the changes needed?

Currently, Spark holds the Future and then cancel them with mayInterruptIfRunning = true in many places.
mayInterruptIfRunning requires the Runnable could response the thread interrupt. There are many places use it in a wrong way.
The correct usage show below.

val future = threadPool.submit(new Runnable {
  override def run(): Unit = try {
    // do something that response the thread interrupt or probably throws InterruptedException.
  } catch {
    case ie: InterruptedException =>
      Thread.currentThread().interrupt()
  }
})

future.cancel(true)

There are two correct example show below.
First, the task support interruption.

Future<?> future = executor.submit(() -> {
    try {
        while (!Thread.currentThread().isInterrupted()) {
            // do something.
        }
    } catch (InterruptedException e) {
        log("Task interrupted, exiting...");
        Thread.currentThread().interrupt(); // restore interrupt state
    }
});

future.cancel(true);

Second, the task body contains operation that could response interrupt.

Future<?> future = executor.submit(() -> {
    try {
        // do something.
        BlockingQueue<String> queue = new LinkedBlockingQueue<>();
        String data = queue.take(); // The block operation supports response interrupt.
        // do something.
    } catch (InterruptedException e) {
        log("Task interrupted, exiting...");
        Thread.currentThread().interrupt(); // restore interrupt state
    }
});

future.cancel(true);

Does this PR introduce any user-facing change?

'No'.

How was this patch tested?

GA

Was this patch authored or co-authored using generative AI tooling?

'No'.

[beliefer]

timeoutCheckingTask = eventLoopThread.scheduleAtFixedRate(
  () => Utils.tryLogNonFatalError { Option(self).foreach(_.ask[Boolean](ExpireDeadHosts)) },
  0, checkTimeoutIntervalMs, TimeUnit.MILLISECONDS)

There is no method called in the task that will throw InterruptdEException.

[beliefer]

      val future = executorService.scheduleAtFixedRate(() => {
        Utils.tryLogNonFatalError(removeIdleFetchedData())
      }, 0, evictorThreadRunIntervalMillis, TimeUnit.MILLISECONDS)
      Some(future)

There is no method called in the task that will throw InterruptdEException.

[srowen]

Why change this though? if it were interruptible, we'd want to interrupt for sure.
What is the issue if you try to interrupt something and it doesn't do anything?

[beliefer]

These tasks can't response interrupt even if we want to interrupt them. cancel(true) causes thread interrupt, but it will not affect these tasks that can't response interrupt. In addition, interrupting operations have extra overhead.

[srowen]

If true, then this change doesn't do anything, right? I wonder what problem this change is solving.

[beliefer]

This PR want avoid the overhead of the invalid thread interruption.

[srowen]

Is there overhead? it just sets the interrupted status of the thread.
I'm worried that, if the code changed in some way that would make it interruptible, then we lose the ability to interrupt on cancel.

[beliefer]

One hand, I want avoid the overhead of the invalid thread interruption, even if the overhead is not big enough.
On the other hand, I suggest the reasonable use of interrupts: use interrupt mechanisms only when needed to avoid excessive dependence.

[beliefer]

      registrationRetryTimer.set(registrationRetryThread.schedule(new Runnable {
        override def run(): Unit = {
          if (registered.get) {
            registerMasterFutures.get.foreach(_.cancel(true))
            registerMasterThreadPool.shutdownNow()
          } else if (nthRetry >= REGISTRATION_RETRIES) {
            markDead("All masters are unresponsive! Giving up.")
          } else {
            registerMasterFutures.get.foreach(_.cancel(true))
            registerWithMaster(nthRetry + 1)
          }
        }
      }, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))

There is no method called in the task that will throw InterruptdEException.

[beliefer]

          recoveryCompletionTask = forwardMessageThread.schedule(new Runnable {
            override def run(): Unit = Utils.tryLogNonFatalError {
              self.send(CompleteRecovery)
            }
          }, recoveryTimeoutMs, TimeUnit.MILLISECONDS)

There is no method called in the task that will throw InterruptdEException.

[beliefer]

    checkForWorkerTimeOutTask = forwardMessageThread.scheduleAtFixedRate(
      () => Utils.tryLogNonFatalError { self.send(CheckForWorkerTimeOut) },
      0, workerTimeoutMs, TimeUnit.MILLISECONDS)

There is no method called in the task that will throw InterruptdEException.

[beliefer]

          registrationRetryTimer = Some(
            forwardMessageScheduler.scheduleAtFixedRate(
              () => Utils.tryLogNonFatalError { self.send(ReregisterWithMaster) },
              PROLONGED_REGISTRATION_RETRY_INTERVAL_SECONDS,
              PROLONGED_REGISTRATION_RETRY_INTERVAL_SECONDS,
              TimeUnit.SECONDS))

ditto

[beliefer]

ditto.

[beliefer]

ditto

[beliefer]

ping @srowen @dongjoon-hyun @LuciferYang