Ingestion Handoff Validation for Apache Druid

The most expensive bug in an automated Druid pipeline is also the most innocuous-looking: the ingestion task reported SUCCESS, so the pipeline moved on, and a downstream query read data that was not there yet. A task's SUCCESS means its segments were persisted to deep storage and published in the metadata store — it does not mean a Historical has loaded them or that the Broker will route a query to them. Those are two distinct events, and the gap between them can be seconds or, when load rules or capacity are wrong, forever. This guide, part of automated ingestion pipeline orchestration, defines the boundary between handoff (persist and publish) and availability (loaded and announced), and shows how to gate downstream work on availability using the Coordinator's serverview, loadstatus, and per-segment used flags.

Mechanics & Internals #

To validate handoff you have to know precisely which component owns each step. When an index_parallel or Kafka indexing task finishes writing a segment, it does two things atomically: it uploads the segment to deep storage, and it inserts a row into the metadata store's segments table with used = 1. At that instant the segment is published — it exists, it is authoritative for its interval, and a fresh cluster restart would load it — but no running Historical is serving it yet. The task then reports SUCCESS and exits. This is handoff in the narrow, technical sense Druid uses: the indexing task has handed the segment off to the metadata store and relinquished responsibility for it. Nothing about it is queryable yet.

Availability is a separate, Coordinator-driven process. The Coordinator runs a periodic management cycle (every druid.coordinator.period, 60 seconds by default) in which it reads the current set of used segments from metadata, evaluates the datasource's load rules to decide how many replicas each segment needs and on which tier, and issues load assignments to Historicals with free capacity. Each assigned Historical pulls the segment from deep storage into its local segment cache and then announces it — publishes its presence to the discovery layer (ZooKeeper or the HTTP-based serverview). The Broker watches that same discovery layer to build its timeline of which server holds which segment; only once a Historical has announced a segment does the Broker add it to the timeline and route matching queries to it. So the full chain from task completion to a query returning the new rows is: publish → Coordinator load assignment → Historical pull → Historical announce → Broker timeline update. Any pipeline that gates on the first arrow alone is reading a promise, not data. This chain is the ingestion-side mirror of query routing and segment discovery, which describes the same timeline from the read side.

Three REST surfaces expose the state of this chain, and a validator uses all three. GET /druid/coordinator/v1/loadstatus returns, per datasource, the percentage of used segments currently loaded across the Druid cluster; 100 (or the datasource absent from the response) means every published segment for that datasource is available. GET /druid/coordinator/v1/loadstatus?full returns the count of segments still pending load per datasource and tier, which is the number you want to watch trend to zero for a specific ingest. The serverview, GET /druid/coordinator/v1/servers?simple or the per-datasource GET /druid/coordinator/v1/datasources/{ds}/segments?full=true, lets you confirm a specific interval's segments are present and carry used: true with non-zero size. The per-segment used flag is the ground truth: a segment can be published-but-unloaded (used: true, not yet on any Historical) or retired (used: false, overshadowed by a newer version). A validator that checks only the aggregate loadstatus percentage can be fooled when many datasources ingest concurrently — the Druid cluster is 100 on average while this interval is still loading — so a robust gate combines the aggregate signal with a per-interval segment check.

Why does the gap ever stretch to never? Three structural reasons. First, a load rule excludes the interval: if a dropByPeriod or a period-bounded loadByPeriod rule does not cover the freshly ingested interval, the Coordinator will never assign it, and the segment stays published-but-unloaded permanently — the rules that govern this are the segment retention policies. Second, no Historical has free capacity: every Historical in the target tier is full, so the load queue backs up and never drains. Third, the segment is overshadowed before it loads: a newer version published in the same cycle wins, and the older one goes straight to unused. A validator must distinguish "still loading, will succeed" from "will never load", which means bounding the wait and surfacing why — pending load versus excluded by rule — rather than polling forever. This gate is the natural completion signal for the retry contract in idempotent ingestion & retry patterns and the sensor condition in workflow orchestrator integration.

Validated Configuration Spec #

Handoff validation is a client-side gate, but it depends on the ingestion spec bounding its interval so the validator knows exactly what to check. The index_parallel spec below is complete and copy-ready; its bounded granularitySpec.intervals is the window the validator polls the Coordinator for.

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "events_web",
      "timestampSpec": { "column": "event_ts", "format": "iso" },
      "dimensionsSpec": {
        "dimensions": ["country", "device", { "name": "page_id", "type": "long" }]
      },
      "metricsSpec": [
        { "type": "count", "name": "rows" },
        { "type": "longSum", "name": "bytes", "fieldName": "resp_bytes" }
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "MINUTE",
        "rollup": true,
        "intervals": ["2026-07-16T00:00:00Z/2026-07-17T00:00:00Z"]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "prefixes": ["s3://lake/events/web/2026-07-16/"] },
      "inputFormat": { "type": "json" },
      "appendToExisting": false,
      "dropExisting": true
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsInMemory": 1000000,
      "maxNumConcurrentSubTasks": 4,
      "forceGuaranteedRollup": true,
      "partitionsSpec": {
        "type": "range",
        "partitionDimensions": ["country"],
        "targetRowsPerSegment": 5000000
      }
    }
  }
}

The pairing that makes the gate work is a loadByPeriod or loadForever rule that actually covers the ingested interval. If the datasource's retention rules are period-bounded, confirm the ingested window falls inside a loadByPeriod and not inside a dropByPeriod, or the segments will publish and never load. The load rule that governs availability is itself configured through POST /druid/coordinator/v1/rules/{datasource}; keeping the ingest interval and the load rules in sync is the precondition for handoff validation to ever pass.

Sizing Heuristics & Formulas #

The validator's central sizing question is: how long should it wait before declaring a handoff failed rather than merely slow? The load latency is bounded below by the Coordinator period and the time to pull the segment bytes from deep storage. If the Coordinator period is ( p ) and a fresh ingest lands ( S ) segments totaling ( B ) bytes onto Historicals whose aggregate pull bandwidth is ( R ), the expected time from publish to fully-available is roughly:

$$ t_{avail} \approx p + \frac{B}{R} $$

The Coordinator term ( p ) is there because publication and the management cycle are unsynchronized — a segment published just after a cycle starts waits nearly a full ( p ) for the next one. For a default ( p = 60\text{s} ), ( B = 1.2\text{ GB} ), and ( R = 400\text{ MB/s} ) of pull bandwidth, ( t_{avail} \approx 60 + 3 = 63\text{s} ). Set the validator's timeout to a comfortable multiple of this — commonly 5 to 10 times — so a busy load queue or a slower deep-storage read does not trip a false failure:

$$ \text{timeout} \approx m \times t_{avail}, \quad m \in [5, 10] $$

The poll interval, meanwhile, should be near the Coordinator period rather than sub-second: polling loadstatus every second when the Coordinator only reassigns every 60 seconds just wastes requests. A poll interval of ( p / 4 ) to ( p / 2 ) catches the transition promptly without hammering the endpoint. Finally, the number of segments a single ingest produces — which sets how much the load queue must drain — follows the interval's row count over the target rows per segment:

$$ S \approx \left\lceil \frac{\text{rowsPerInterval}}{\text{targetRowsPerSegment}} \right\rceil $$

so a fine segmentGranularity that fragments an interval into many small segments lengthens the load-queue drain and therefore the handoff wait, one more reason streaming output is later consolidated by automated compaction task scheduling.

Python Orchestration Snippet #

The validator below gates on availability using all three signals — aggregate loadstatus, per-interval segment presence, and the used flag — with capped exponential backoff. It uses only the standard library plus requests, and returns a structured result so the caller can distinguish "loaded" from "still pending" from "timed out".

import time
import requests

COORDINATOR = "http://coordinator:8081"


def loadstatus_pct(datasource: str) -> float:
    """Fraction of used segments loaded cluster-wide for this datasource."""
    r = requests.get(f"{COORDINATOR}/druid/coordinator/v1/loadstatus", timeout=15)
    r.raise_for_status()
    body = r.json()
    # A datasource absent from loadstatus is fully loaded (100%).
    return float(body.get(datasource, 100.0))


def interval_segments(datasource: str, interval_start: str) -> list:
    """Used segments whose interval falls in the target window, with real bytes."""
    r = requests.get(
        f"{COORDINATOR}/druid/coordinator/v1/datasources/{datasource}/segments",
        params={"full": "true"}, timeout=15)
    r.raise_for_status()
    return [s for s in r.json()
            if s.get("interval", "").startswith(interval_start)
            and s.get("used", False) and s.get("size", 0) > 0]


def await_availability(datasource: str, interval_start: str,
                       timeout_s: int = 600) -> dict:
    """Gate on availability, not SUCCESS. Return a structured verdict."""
    deadline = time.monotonic() + timeout_s
    delay = 15.0
    while time.monotonic() < deadline:
        pct = loadstatus_pct(datasource)
        segs = interval_segments(datasource, interval_start)
        if pct >= 100.0 and segs:
            return {"available": True, "segments": len(segs),
                    "version": sorted({s["version"] for s in segs})[-1]}
        time.sleep(delay)
        delay = min(delay * 1.5, 30.0)   # gentle backoff near the Coordinator period
    # Timed out: report whether it is pending-load or never-assigned.
    segs = interval_segments(datasource, interval_start)
    return {"available": False, "loadstatus_pct": loadstatus_pct(datasource),
            "published_segments": len(segs),
            "hint": "published but unloaded — check load rules / Historical capacity"
                    if segs else "no published segments — task may not have handed off"}


def gate_downstream(datasource: str, interval_start: str) -> None:
    verdict = await_availability(datasource, interval_start)
    if not verdict["available"]:
        raise RuntimeError(f"handoff not validated: {verdict}")

Three properties make this a trustworthy gate. Two-signal confirmation: it requires both loadstatus at 100% and a concrete per-interval segment with used: true and non-zero size, so a concurrent datasource cannot make the aggregate lie about this interval. Actionable timeout: on failure it distinguishes "published but unloaded" (a load-rule or capacity problem) from "no published segments" (the task never handed off), so the on-call gets a root-cause hint, not just a timeout. Coordinator-aware backoff: the poll interval sits near the Coordinator period instead of spinning sub-second. The concrete curl-level walkthrough of these same checks lives in verifying segment handoff completion.

Failure Modes & Diagnostics #

Handoff-validation failures split cleanly by which signal disagrees.

1. Task SUCCESS, query returns nothing. The pipeline gated on the wrong event. Confirm the segments are published but not loaded:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadstatus?full" | jq '.events_web'

A non-empty per-tier pending count means load is still in progress; zero-but-not-queryable means a rule excludes it.

2. loadstatus sits below 100% and never reaches it. Either no Historical has capacity or a rule blocks the interval. Check the load queue and the assignable capacity:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadqueue?simple" \
  | jq 'to_entries[] | {server: .key, toLoad: (.value.segmentsToLoad | length)}'

A load queue that never drains points at full Historicals — add capacity or adjust the tier's replication.

3. Segments show used: false immediately after ingest. A newer version overshadowed them, or a dropByPeriod retired them. Inspect versions and used flags for the interval:

curl -s "http://coordinator:8081/druid/coordinator/v1/datasources/events_web/segments?full=true&includeUnused=true" \
  | jq '[.[] | select(.interval | startswith("2026-07-16"))]
        | group_by(.version)[] | {version: .[0].version, used: .[0].used}'

4. Aggregate loadstatus is 100% but this interval is not queryable. The aggregate was satisfied by other intervals. This is exactly why the validator also checks per-interval segments; confirm the specific window:

curl -s "http://coordinator:8081/druid/coordinator/v1/datasources/events_web/segments?full=true" \
  | jq '[.[] | select(.interval | startswith("2026-07-16T13"))] | length'

Zero here despite 100% aggregate means the interval genuinely has no loaded segments.

5. Segments load, then disappear minutes later. A retention rule dropped them after load — the interval fell outside a loadByPeriod window. Reconcile the ingest interval against the datasource's rules via configuring segment retention policies.

Automation Checklist #

Wire these into the completion gate of every ingestion job.

• Never complete on task SUCCESS alone — treat SUCCESS as "published", then gate on availability.
• Require two signals — both loadstatus at 100% and a concrete per-interval segment with used: true and non-zero size.
• Check the specific interval — a per-interval segment query, not just the aggregate, so concurrent datasources cannot mask a pending load.
• Bound the wait — set a timeout at 5–10× the expected load latency; do not poll forever.
• Classify a timeout — distinguish "published but unloaded" (rules/capacity) from "never handed off" (task problem) in the failure message.
• Poll near the Coordinator period — a poll interval of p/4–p/2, not sub-second.
• Reconcile ingest interval with load rules — confirm a loadByPeriod/loadForever covers the window and no dropByPeriod excludes it.
• Watch the load queue — alert when loadqueue for a tier stops draining, the signature of full Historicals.
• Confirm a single live version — after a replace, exactly one used version should cover the interval.
• Emit availability latency — record publish-to-available time per ingest so a lengthening handoff is caught before it breaches SLA.

Related #

• Verifying segment handoff completion — the concrete curl-level checks against serverview and loadstatus that confirm queryability.
• Workflow orchestrator integration — make this availability gate the sensor condition in an Airflow or Dagster DAG.
• Idempotent ingestion & retry patterns — pair handoff validation with content-hashed ids so a retried task's segments are the ones you confirm.
• Query routing and segment discovery — the read-side view of the same Broker timeline that availability feeds.
• Automated compaction task scheduling — consolidate the fragmented segments whose count lengthens the load-queue drain.

Up one level: Automated ingestion pipeline orchestration frames how handoff validation, idempotent submission, and orchestration compose into one deterministic pipeline.