Druid Segment Ops Segment & Ingestion Orchestration

Non-Blocking Druid Task Status Polling

Submit fifty backfill tasks and the naive supervisor loops over them one at a time — polling task one to completion before it ever checks task two — so a single slow task stalls the status of every task behind it, and a run that should take minutes takes as long as the sum of its parts. The fix is to poll all in-flight tasks concurrently, each on its own capped-backoff schedule, so the loop advances at the pace of the slowest single task rather than their sum. This guide builds that non-blocking poller with concurrent.futures from the standard library, reconciling many task states in one pass. It is the fan-out companion to the submit-poll-complete model in async task execution patterns.

Sequential versus concurrent polling of many in-flight tasks In sequential polling, tasks are checked one after another, so total time is the sum of each task's wait and one slow task blocks the rest. In concurrent polling, a bounded thread pool polls every task at once on its own capped-backoff schedule, so total time is bounded by the single slowest task and results are reconciled together. Sequential — total = sum of waits task 1 task 2 (slow) — blocks rest task 3 task 4 time Concurrent — total ≈ slowest single task task 1 task 2 (slow) task 3 task 4 time all done
Concurrent polling bounds wall-clock time by the slowest task, not the sum of all tasks.

Failure Modes & Diagnostics

The failures here are throughput failures, not correctness failures: the states are right, but the loop is too slow or too aggressive. A sequential poller's wall-clock time grows linearly with task count; a concurrent poller that hammers the Overlord with a tight loop and no backoff overwhelms it during a leader election. Between those extremes, tasks that finish but are never reaped leave the poller waiting forever. Start by listing what is actually in flight and how the Overlord is coping:

# All running tasks and their datasource — the working set to poll
curl -s "http://<overlord-host>:8090/druid/indexer/v1/tasks?state=running" \
  | jq 'map({id: .id, ds: .dataSource, created: .createdTime})'

# Status of a single task (the call the poller makes per task)
curl -s "http://<overlord-host>:8090/druid/indexer/v1/task/<taskId>/status" \
  | jq '.status.status'   # RUNNING | SUCCESS | FAILED

# Worker capacity — a full pool means new tasks sit PENDING, not RUNNING
curl -s "http://<overlord-host>:8090/druid/indexer/v1/workers" \
  | jq 'map({host: .worker.host, used: .currCapacityUsed, cap: .worker.capacity})'

If most tasks show PENDING rather than RUNNING, the bottleneck is worker capacity, not the poller — the submit rate outran available slots, a pattern triaged in debugging Druid supervisor task failures. If the Overlord returns 429s or times out, the poll concurrency or frequency is too high; widen the backoff. The submit side that produces this working set — deterministic IDs and capped-backoff submission — is covered in dynamic ingestion spec generation.

Target Spec & Validated JSON

There is no ingestion spec to tune here; the "spec" being validated is the poller's own contract — its concurrency bound, per-task backoff schedule, and terminal-state set. Encode it as an explicit config so the poll behaviour is reviewable rather than buried in constants:

{
  "poller": {
    "maxConcurrency": 16,
    "perTaskTimeoutSeconds": 3600,
    "overallDeadlineSeconds": 7200,
    "backoff": {
      "baseSeconds": 2,
      "capSeconds": 30,
      "multiplier": 2
    },
    "terminalStates": ["SUCCESS", "FAILED", "INTERRUPTED"]
  }
}

The load-bearing settings:

  • maxConcurrency — the size of the thread pool, and therefore the ceiling on simultaneous in-flight requests to the Overlord. Bound it well below druid.worker.capacity so status polling never competes with task submission for the Overlord's request threads. The number of concurrent connections a run holds is $\min(\text{maxConcurrency}, \text{numTasks})$.
  • backoff — the per-task schedule. Each task starts at baseSeconds and doubles to capSeconds, so a task that finishes quickly is reaped after one or two short waits while a long task settles into a steady polite cadence instead of a busy-loop.
  • terminalStates — the states that stop polling a task. Every non-terminal state (RUNNING, PENDING, WAITING) means keep polling; anything in this set means reap and record.

The wall-clock time of a run over $N$ tasks with concurrency $c$ and a slowest task duration $T_{\max}$ is bounded below by the concurrency ceiling and the slowest task:

$$ T_{\text{run}} \approx \max!\left(T_{\max},\ \left\lceil \frac{N}{c} \right\rceil \times \bar{t} \right) $$

where $\bar{t}$ is the average time a task occupies a worker slot. When $c \ge N$, the run collapses to $\approx T_{\max}$ — the whole point of polling concurrently.

Python Automation Script

The poller below uses concurrent.futures.ThreadPoolExecutor from the standard library plus requests. Each task is polled on its own thread with capped exponential backoff until it reaches a terminal state or its per-task deadline; as_completed reaps results as they finish so the caller sees progress without waiting for the whole batch. I/O-bound polling is a textbook fit for a thread pool — the threads spend nearly all their time blocked on the network, so the GIL is not a bottleneck.

import time
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

OVERLORD = "http://overlord:8090"
TERMINAL = {"SUCCESS", "FAILED", "INTERRUPTED"}


def poll_one(task_id: str, per_task_timeout: float = 3600.0,
             base: float = 2.0, cap: float = 30.0) -> tuple[str, str]:
    """Poll a single task to a terminal state with capped exponential backoff."""
    delay, waited = base, 0.0
    session = requests.Session()
    while waited < per_task_timeout:
        try:
            r = session.get(
                f"{OVERLORD}/druid/indexer/v1/task/{task_id}/status", timeout=15
            )
            r.raise_for_status()
            status = r.json()["status"]["status"]
        except requests.RequestException:
            status = None  # transient; fall through to backoff and retry
        if status in TERMINAL:
            return task_id, status
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, cap)  # capped backoff, per task
    return task_id, "TIMEOUT"


def poll_all(task_ids: list[str], max_concurrency: int = 16,
             overall_deadline: float = 7200.0) -> dict[str, str]:
    """Poll many tasks concurrently; reap each as it reaches a terminal state."""
    results: dict[str, str] = {}
    start = time.monotonic()
    workers = min(max_concurrency, len(task_ids)) or 1
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(poll_one, tid): tid for tid in task_ids}
        for future in as_completed(futures):
            tid = futures[future]
            results[tid] = future.result()[1]
            elapsed = time.monotonic() - start
            print(f"[{elapsed:6.1f}s] {tid} -> {results[tid]}")
            if elapsed > overall_deadline:
                # Record the unfinished tasks and stop waiting.
                for pending, ptid in futures.items():
                    results.setdefault(ptid, "DEADLINE")
                break
    return results


def run(task_ids: list[str]) -> int:
    results = poll_all(task_ids)
    failed = {t: s for t, s in results.items() if s != "SUCCESS"}
    if failed:
        print(f"NON-SUCCESS TASKS: {failed}")
        return 1
    print(f"ALL {len(results)} TASKS SUCCEEDED")
    return 0

Three properties keep this safe under load. Bounded concurrency: the pool size caps simultaneous Overlord requests, so a batch of 500 tasks never opens 500 connections. Per-task backoff: each thread backs off independently, so a task that lingers in RUNNING polls at the capped cadence while a quick task is reaped after one short wait — no busy-loop, no thundering herd. Deadlines at both levels: a per-task timeout stops an individual runaway, and an overall deadline stops the whole run from hanging on a stuck task, recording the unfinished ones rather than blocking forever.

Verification Steps

Confirm the poller reaps all tasks and bounds its wall-clock time. Submit a small batch, capture their IDs, and run the poller:

python3 -c "from poller import run; import sys; sys.exit(run(open('task_ids.txt').read().split()))"
echo "exit=$?"
# [  4.0s] index_parallel_events_a -> SUCCESS
# [ 12.0s] index_parallel_events_b -> SUCCESS
# [ 30.0s] index_parallel_events_c -> SUCCESS
# ALL 3 TASKS SUCCEEDED
# exit=0

The timestamps confirm concurrency: task C finishing at 30s while A finished at 4s means they were polled in parallel, not in series (a sequential poller would show C at 4+12+30 = 46s). Cross-check each reported state against the Overlord directly:

for t in $(cat task_ids.txt); do
  echo "$t: $(curl -s "http://<overlord-host>:8090/druid/indexer/v1/task/$t/status" | jq -r '.status.status')"
done
# index_parallel_events_a: SUCCESS
# index_parallel_events_b: SUCCESS
# index_parallel_events_c: SUCCESS

Confirm the poller never overwhelmed the Overlord — worker and request metrics should show no 429 spike during the run:

curl -s "http://<overlord-host>:8090/druid/indexer/v1/workers" \
  | jq 'map({host: .worker.host, used: .currCapacityUsed, cap: .worker.capacity})'

Wire the poller as the wait step after a fan-out submit: keep maxConcurrency below worker capacity, treat any non-SUCCESS state as a failed run, and export the results map so downstream steps see exactly which task ids succeeded. Reference the Druid tasks API documentation for the full status and report payload shapes.

Up one level: Async task execution patterns frames how non-blocking submission and polling keep long-running ingestion supervisable.

Back to Automated Ingestion Pipeline Orchestration