apache / apache/texera

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

Open
#6,923 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Scala
Stars
314
Forks
187
Avg merge
1d 21h
Merged PRs (30d)
214

Description

### 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:

```scala
// 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 429be110a7; discovered during the investigation for #6916).

Contributor guide

Open the contributing guide

Research direction

Start in RegionExecutionManager.scala at terminateWorkersWithRetry and its JavaTimer retry continuation, then inspect AsyncRPCClient.scala and NetworkOutputGateway.scala at the listed shared-state declarations and mutations. Trace the coordinator actor and timer-thread paths, including gracefulStop callbacks. Done means retries no longer concurrently mutate coordinator RPC state, with the relevant behavior verified by tests or inspection.

Written by the indexing model from the issue text.

Assessment

Tech stack
scala
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.