Region-termination retries run on a JavaTimer thread and mutate unsynchronized coordinator RPC state

[aglinxinyuan]

What happened?

Region-termination retry attempts run on a JavaTimer thread and mutate coordinator RPC state that is not thread-safe, concurrently with the coordinator's actor thread. This is a latent data race — no observed corruption yet, but nothing prevents it.

Thread switch: attempt 1 of terminateWorkersWithRetry runs on the coordinator actor thread (inside the portCompleted handler continuation), but every retry runs on the timer thread:

// RegionExecutionManager.scala:129
private val killRetryTimer: Timer = new JavaTimer(true)
// :255-257
Future.sleep(killRetryDelay)(killRetryTimer)
  .flatMap(_ => terminateWorkersWithRetry(regionExecution, attempt + 1))   // <- runs on the JavaTimer thread

terminateWorkers then calls asyncRPCClient.workerInterface.endWorker(...), which touches unsynchronized shared state:

state	declaration	mutated by
promiseID	plain var (AsyncRPCClient.scala:137)	createPromise (:145-150) on the timer thread; same path on the actor thread for every other RPC
unfulfilledPromises	unsynchronized mutable.HashMap (AsyncRPCClient.scala:136)	createPromise on the timer thread; fulfillPromise on the actor thread for every inbound reply
idToSequenceNums	unsynchronized mutable.HashMap (NetworkOutputGateway.scala:47)	getSequenceNumber uses getOrElseUpdate (:93-95) with no lock (note removeControlChannel :97-101 IS synchronized, so the intent to synchronize exists but is inconsistent)

Consequences range from lost/duplicated promise ids (an RPC whose reply can never be matched → termination hang) to HashMap structural corruption under concurrent resize.

Also on this path: gracefulStop(...) future callbacks update WorkerExecution state from Pekko dispatcher threads (RegionExecutionManager.scala:214-219), so worker-state writes are concurrent with coordinator-thread reads (this interacts with the stats-aggregation issue, filed separately).

Fix directions: hop back to the coordinator's serialized execution context for the retry continuation (instead of running terminateWorkers directly on the timer thread), or make the touched structures thread-safe and document the invariant.

How to reproduce?

Code inspection; the race window is every retried termination attempt (any teardown that fails attempt 1) racing against any concurrent coordinator RPC (e.g. periodic stats queries).

Version/Branch

main (observed at 429be11; discovered during the investigation for #6916).