aho135 opened a new issue, #20339:
URL: https://github.com/apache/druid/issues/20339
### Affected Version
Observed on 36.x; the relevant code paths are unchanged on `master` (as of
commit `2bf1643cdd`). Uses the HTTP-based task runner
(`druid.indexer.runner.type=httpRemote`).
### Summary
If a MiddleManager's node-discovery entry **flaps (removed then re-added)
during the Overlord's startup worker-sync**, the Overlord's `becomeLeader()`
blocks for the full `3 × druid.indexer.runner.syncRequestTimeout` (default `3 ×
PT3M = 9 minutes`) awaiting a stale `WorkerHolder` whose initialization latch
can never fire, then throws and **fails leadership acquisition**. During that
window the Overlord is effectively leaderless (no supervisor management), and
the failure triggers a further leadership transition.
### Description
**Cluster / config**
- `httpRemote` task runner; combined Coordinator/Overlord process.
- ~12 MiddleManagers, ~40 Kafka streaming supervisors.
- `druid.indexer.runner.syncRequestTimeout` left at its default `PT3M`.
- ZooKeeper-based discovery (`CuratorDruidNodeDiscoveryProvider`).
**Trigger**
The ZooKeeper ensemble leader restarted, causing a brief re-election and
Curator `SUSPENDED → RECONNECTED` on the Druid processes. This forced an
Overlord leadership change. As the new leader ran `becomeLeader()`, one
MiddleManager's ZK session had also reset, so its ephemeral discovery node was
deleted and immediately re-created — a remove+add flap occurring within ~1s of
the Overlord starting its worker sync.
**Observed behavior (timestamps relative)**
```
T+0.000 New leader: HttpRemoteTaskRunner - "[12] Workers are discovered."
T+0.000 ChangeRequestHttpSyncer - "Starting sync for server[...MM-X...]"
(x12)
T+0.000 HttpRemoteTaskRunner - "Waiting for worker[MM-0] to sync state..."
T+1.017 HttpRemoteTaskRunner - "Kaboom! Worker[MM-X] removed!" # old
ZK node gone
T+1.350 HttpRemoteTaskRunner - "Worker[MM-X] reportin' for duty!" # new ZK
node -> new holder, syncs fine
T+1.359 HttpRemoteTaskRunner - "Task[...] location changed on worker[MM-X]"
# new holder healthy
...
T+555.4 ERROR CuratorDruidLeaderSelector - "listener becomeLeader() failed.
Unable to become leader"
```
Stack trace (host scrubbed):
```
java.lang.RuntimeException: java.lang.RuntimeException:
org.apache.druid.java.util.common.RE:
Failed to sync with worker[<middlemanager-host>:8088].
at
org.apache.druid.indexing.overlord.DruidOverlord$1.becomeLeader(DruidOverlord.java)
...
```
The block lasted ~555s ≈ `3 × PT3M` (540s) plus the time to await the
earlier, healthy workers. The re-added MiddleManager was healthy the entire
time (its new `WorkerHolder` synced within ~1s and streamed task snapshots);
only the **stale, stopped holder** the startup loop was awaiting never
initialized.
### Root cause
Leadership callbacks run on a single-threaded executor and `becomeLeader()`
is invoked synchronously, so the whole node cannot process any further
leadership transition until it returns:
- `CuratorDruidLeaderSelector#createNewLeaderLatchWithListener` —
`isLeader()` calls `listener.becomeLeader()` inline on the
`LeaderSelector[...]` single-thread executor; on any thrown exception it alerts
and calls `notLeader()`.
`becomeLeader()` starts the task runner (a managed lifecycle instance)
**before** the supervisor manager, and `HttpRemoteTaskRunner.start()` blocks
until it has synced with every discovered worker:
- `DruidOverlord$1.becomeLeader()` builds the `"task-master"` `Lifecycle`,
adds `taskRunner` (`HttpRemoteTaskRunner`) as a managed instance, then
`supervisorManager`, then `leaderLifecycle.start()`.
- `HttpRemoteTaskRunner#startWorkersHandling()` — after worker discovery,
iterates `for (WorkerHolder worker : workers.values()) {
worker.waitForInitialization(); }`. The loop holds `WorkerHolder` references
obtained at loop start.
The per-worker wait can never complete for a flapped worker:
- `WorkerHolder#waitForInitialization()` → `syncer.awaitInitialization()`;
returns `false` → `throw new RE("Failed to sync with worker[%s]")`.
- `ChangeRequestHttpSyncer`: `maxDurationToWaitForSync = 3 *
serverHttpTimeout` and `awaitInitialization()` awaits `initializationLatch` up
to that duration. `serverHttpTimeout` is
`HttpRemoteTaskRunnerConfig#getSyncRequestTimeout()` (default `PT3M`).
**`initializationLatch` is only counted down on the first successful full/delta
sync.**
- When the worker's discovery node is removed,
`HttpRemoteTaskRunner#removeWorker()` does `workers.remove(host)` then
`workerHolder.stop()` → `syncer.stop()`. **Stopping the syncer does not count
down `initializationLatch`.**
So when the node flaps mid-startup: the old holder the startup loop is
awaiting gets `stop()`ped (latch stuck at 1 forever), while a brand-new holder
is created via `addWorker()` and syncs normally. The loop keeps awaiting the
**dead** holder, blocks the full `3 × syncRequestTimeout`, then throws —
failing `becomeLeader()`.
### Impact
- Overlord leadership acquisition is delayed/failed for up to `3 ×
druid.indexer.runner.syncRequestTimeout` (default **9 minutes**) whenever a
worker's discovery entry flaps during that startup window — likely during the
very ZK instability that caused the leadership change in the first place.
- During the gap the Overlord never reaches `SupervisorManager.start()`, so
streaming supervisors are unmanaged: Kafka indexing tasks that reach their
`taskDuration` and exit are not replaced, and ingestion falls behind. When
leadership finally settles on another node, all supervisors are (re)started at
once, producing a synchronized ingestion-lag spike across every datasource on
the cluster.
- The `becomeLeader()` failure calls `notLeader()`, adding another
leadership transition to an already-unstable moment.
### Suggested fixes (for discussion)
1. In `startWorkersHandling()`, don't await a captured `WorkerHolder`
reference — re-check the current holder in `workers` (skip/replace ones that
have been removed/stopped), or await by host and drop the wait if the entry
disappears.
2. Have `WorkerHolder#stop()` / `ChangeRequestHttpSyncer#stop()` count down
(or otherwise release) `initializationLatch` so `awaitInitialization()` returns
promptly for a stopped syncer instead of blocking the full timeout.
3. Make the initial worker sync best-effort: log-and-continue when a single
worker can't sync at startup (it will sync via the normal `addWorker` path)
rather than failing the entire `becomeLeader()` — optionally bound the
aggregate wait rather than paying `3 × syncRequestTimeout` per worker.
### Debugging already done
Correlated Overlord logs (`becomeLeader` start → `becomeLeader() failed`
after ~555s with `Failed to sync with worker[...]`), the affected
MiddleManager's logs (process healthy throughout; ZK node re-announced;
successfully POSTing to the new Overlord), and the discovery flap on the
Overlord (`Kaboom! Worker[...] removed!` then `reportin' for duty!` within
~1s), against the code paths above. Timing matches `3 × PT3M`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
