Overlord becomeLeader() blocks for 3 × druid.indexer.runner.syncRequestTimeout and fails leadership acquisition if a MiddleManager's discovery entry flaps during startup worker sync #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.