An End-to-End Ingestion CI/CD Gate for Druid Segments

A production ingestion change to Apache Druid — a new dimension, a retuned partitionsSpec, a fresh backfill window — is only as safe as the checks that run before it is declared shipped. This guide assembles four independent safeguards into a single deploy gate that a CI/CD pipeline runs as one command: validate the assembled spec, submit it and confirm the segments actually become queryable, audit the resulting segment sizes, and verify the monitoring surface is clean before the job exits zero. The gate is deliberately a chain — each stage is a hard boundary that returns a non-zero exit code on failure, so a broken schema never reaches the Overlord, a task that succeeds but never hands off never signals success downstream, and a size regression never silently ships to a Historical. It sits inside the broader segment architecture and lifecycle fundamentals, orchestrating the individual guides that each own one stage in depth into one executable pass/fail decision.

Mechanics & Internals #

The gate treats an ingestion deploy the way a software CI pipeline treats a code change: a sequence of stages that must each pass, run in a fixed order, with the cheapest and most deterministic checks first so a broken change fails fast before it consumes cluster resources. The ordering is not cosmetic. Stage 1 runs entirely offline against the assembled JSON and costs nothing on the Druid cluster; stage 2 consumes an Overlord slot and real ingestion time; stage 3 reads back what stage 2 produced; stage 4 reads the control-plane's own view of the result. Reordering them would waste an Overlord slot on a spec that a millisecond of local validation would have rejected.

Stage 1 — schema validation. Before any network call to the Druid cluster, the assembled spec is checked against a structural contract and a set of Druid-specific semantic rules. This is the same discipline documented under schema validation for Druid specs: confirm the required top-level keys (type, and inside spec the dataSchema, ioConfig, and tuningConfig) are present and well-typed, and that combinations Druid will reject at submission — forceGuaranteedRollup with a dynamic partitionsSpec, or a missing intervals when dropExisting is set — are caught locally. A spec that a generator produced at runtime, as in dynamic ingestion spec generation, still passes through this same gate; generation and validation are separate concerns by design.

Stage 2 — ingestion submit and handoff. The gate POSTs the validated spec to the Overlord at POST /druid/indexer/v1/task, polls GET /druid/indexer/v1/task/{taskId}/status to a terminal state with capped backoff, and then — critically — does not stop at SUCCESS. Task success means the segment bytes were persisted to deep storage and the descriptor was committed to the metadata store; it does not mean the segment is queryable. Handoff (persist + publish) and availability (Coordinator-assigned, Historical-loaded, Broker-announced) are distinct events, and gating downstream work on the wrong one is the single most common "the job passed but the data isn't there" incident. The gate therefore polls the Coordinator serverview until at least one used segment for the target interval is loaded — the discipline owned by ingestion handoff validation.

Stage 3 — post-ingest compaction audit. Once segments are queryable, the gate reads their size distribution back from the Coordinator and asserts they sit inside the target band. A change that quietly produces a swarm of tiny segments or a few oversized ones is a regression even when every task succeeded, because it degrades the control plane and the Historical heap over time. If the distribution has drifted, the gate can trigger a compaction pass — the mechanism formalized in automated compaction task scheduling — and re-audit, or simply fail and hand the deploy back for a partitioning fix. The sizing target itself is the one established in optimizing segment size for Historical nodes.

Stage 4 — monitoring and alert verification. The final stage confirms the Druid cluster's own telemetry agrees the deploy is clean: no segments are unavailable or under-replicated for the affected datasource, and no lifecycle alert is firing. This closes the loop — a task can pass, hand off, and produce correctly sized segments, yet still leave the datasource under-replicated because a target tier is at capacity. Reading the Coordinator's loadstatus and the datasource's segment view turns "the pipeline ran" into "the data is served, replicated, and correctly shaped."

Because the stages are a chain, the gate's contract is simple: exit 0 means all four passed and the change is safe to mark deployed; any non-zero exit names the stage that failed and why, so the CI log points straight at the fix.

Validated Configuration Spec #

The gate operates on a normal index_parallel spec — the artifact under version control that a change actually modifies. Below is the exact shape the gate validates in stage 1 and submits in stage 2, with every required top-level key present so it is copy-ready against the latest stable Druid release. The gate does not invent this spec; it consumes whatever the pipeline produced and holds it to the contract.

{
  "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-02-10T00:00:00Z/2026-02-11T00:00:00Z"]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "prefixes": ["s3://lake/events/web/2026-02-10/"] },
      "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 fields the gate inspects most closely are the ones whose combinations Druid rejects at submission or that determine segment shape:

• granularitySpec.intervals — must be present and must bound exactly the window being built. Combined with ioConfig.dropExisting: true, an explicit intervals makes the task atomically replace every existing segment in that window on handoff, the correct primitive for a deterministic, idempotent deploy. A missing intervals under dropExisting is a stage-1 rejection.
• tuningConfig.forceGuaranteedRollup — requires a non-dynamic partitionsSpec (range or hashed) and a bounded intervals. The gate checks this pairing locally rather than letting the Overlord reject the task after a slot is consumed.
• tuningConfig.partitionsSpec.targetRowsPerSegment — the primary lever the stage-3 audit holds accountable; it must be sized so produced segments land in the target band.
• dataSchema.dimensionsSpec.dimensions — typed dimensions are emitted as objects ({"name": "page_id", "type": "long"}) so Druid does not misinfer them as strings. Full field semantics are maintained in the official Apache Druid ingestion spec reference.

Sizing Heuristics & Formulas #

The gate's stage-3 audit needs a concrete pass/fail boundary, and that boundary is derived, not guessed. Druid's operational sweet spot is roughly 300–700 MB per segment, about 5 million rows for typical event data. Given a target size and the average on-disk row width measured from real segment telemetry, the row target the spec should carry is:

$$ \text{targetRows} \approx \frac{\text{targetMB} \times 1048576}{\text{avgRowBytes}} $$

For a 600 MB target over rows averaging $\approx 120$ bytes on disk, that yields $\approx 5.24 \times 10^{6}$ rows — the targetRowsPerSegment the validated spec carries. The audit does not re-derive this; it reads back the produced sizes and asserts the distribution sits inside the band. A useful pass condition is that the fraction of segments outside $[300, 750]$ MB stays below a small tolerance $\tau$:

$$ \frac{\lvert { s : \text{size}(s) < 300,\text{MB} \ \lor\ \text{size}(s) > 750,\text{MB} } \rvert}{N} \le \tau $$

with $\tau \approx 0.10$ a common starting tolerance. The number of segments an interval produces — which the gate also sanity-checks, since a sudden count spike is the earliest fragmentation signal — follows from the post-rollup row count and the row target:

$$ \text{segmentsPerInterval} \approx \left\lceil \frac{\rho \times \text{rawEventsPerInterval}}{\text{targetRows}} \right\rceil $$

where $\rho$ is the observed rollup ratio (output rows over input rows) from prior runs. Because rollup collapses rows, the gate must always feed these formulas post-rollup telemetry, never the raw event count. The stage-3 pass/fail decision on the size band is worked through in full in the companion guide, blocking deploys on segment size regression.

Python Orchestration Snippet #

The reference gate below runs the four stages in order using only the Python standard library plus requests, with capped exponential backoff on every cluster call. It returns a non-zero process exit on the first stage that fails, printing the stage and reason so a CI log points straight at the problem. It is deliberately linear and side-effect-light: each stage is a function that either returns cleanly or raises GateFailure, and main maps that to an exit code.

import sys
import json
import time
import requests

OVERLORD = "http://overlord:8090"
COORDINATOR = "http://coordinator:8081"

REQUIRED_TOP = ("type", "spec")
REQUIRED_SPEC = ("dataSchema", "ioConfig", "tuningConfig")
SIZE_MIN_MB, SIZE_MAX_MB, OUT_OF_BAND_TOLERANCE = 300, 750, 0.10


class GateFailure(Exception):
    """Raised by any stage to block the deploy with a specific reason."""


def _get_json(url, timeout=30, max_wait=0.0):
    """GET with capped exponential backoff; returns parsed JSON or raises."""
    delay, waited = 2.0, 0.0
    while True:
        try:
            r = requests.get(url, timeout=timeout)
            r.raise_for_status()
            return r.json()
        except requests.RequestException:
            if waited >= max_wait:
                raise
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, 30.0)  # capped backoff


def stage1_validate(spec: dict) -> None:
    for key in REQUIRED_TOP:
        if key not in spec:
            raise GateFailure(f"stage1: missing top-level key '{key}'")
    for key in REQUIRED_SPEC:
        if key not in spec.get("spec", {}):
            raise GateFailure(f"stage1: missing spec.{key}")
    tuning = spec["spec"]["tuningConfig"]
    gran = spec["spec"]["dataSchema"].get("granularitySpec", {})
    if tuning.get("forceGuaranteedRollup"):
        if tuning.get("partitionsSpec", {}).get("type") == "dynamic":
            raise GateFailure("stage1: forceGuaranteedRollup needs non-dynamic partitionsSpec")
        if not gran.get("intervals"):
            raise GateFailure("stage1: forceGuaranteedRollup needs bounded intervals")
    if spec["spec"]["ioConfig"].get("dropExisting") and not gran.get("intervals"):
        raise GateFailure("stage1: dropExisting requires an explicit intervals bound")


def stage2_ingest(spec: dict, datasource: str, interval: str,
                  task_wait: float = 3600.0, avail_wait: float = 600.0) -> str:
    r = requests.post(f"{OVERLORD}/druid/indexer/v1/task", json=spec, timeout=30)
    r.raise_for_status()
    task_id = r.json()["task"]

    delay, waited = 2.0, 0.0
    while waited < task_wait:
        status = _get_json(
            f"{OVERLORD}/druid/indexer/v1/task/{task_id}/status"
        )["status"]["status"]
        if status == "SUCCESS":
            break
        if status in ("FAILED", "INTERRUPTED"):
            raise GateFailure(f"stage2: task {task_id} terminal state {status}")
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, 60.0)  # capped backoff
    else:
        raise GateFailure(f"stage2: task {task_id} did not finish within {task_wait}s")

    # Handoff succeeded; now confirm queryability before signalling success.
    delay, waited = 2.0, 0.0
    iv = interval.replace("/", "_")
    while waited < avail_wait:
        view = _get_json(
            f"{COORDINATOR}/druid/coordinator/v1/datasources/{datasource}"
            f"/intervals/{iv}/serverview"
        )
        if view:
            return task_id
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, 30.0)  # capped backoff
    raise GateFailure(f"stage2: task {task_id} succeeded but segments never became queryable")


def stage3_compaction_audit(datasource: str) -> None:
    segs = _get_json(
        f"{COORDINATOR}/druid/coordinator/v1/datasources/{datasource}/segments?full=true"
    )
    if not segs:
        raise GateFailure("stage3: no segments returned for datasource")
    sizes = [s.get("size", 0) / 1048576 for s in segs]
    out = [m for m in sizes if m < SIZE_MIN_MB or m > SIZE_MAX_MB]
    if len(out) / len(sizes) > OUT_OF_BAND_TOLERANCE:
        raise GateFailure(
            f"stage3: {len(out)}/{len(sizes)} segments outside "
            f"{SIZE_MIN_MB}-{SIZE_MAX_MB} MB band"
        )


def stage4_monitor(datasource: str) -> None:
    load = _get_json(f"{COORDINATOR}/druid/coordinator/v1/loadstatus?full")
    ds_load = load.get(datasource, {})
    unavailable = sum(v for v in ds_load.values()) if isinstance(ds_load, dict) else 0
    if unavailable:
        raise GateFailure(f"stage4: {unavailable} segments still unavailable for {datasource}")


def main() -> int:
    spec = json.load(open(sys.argv[1]))
    datasource = spec["spec"]["dataSchema"]["dataSource"]
    interval = spec["spec"]["dataSchema"]["granularitySpec"]["intervals"][0]
    try:
        stage1_validate(spec)
        task_id = stage2_ingest(spec, datasource, interval)
        stage3_compaction_audit(datasource)
        stage4_monitor(datasource)
    except GateFailure as exc:
        print(f"GATE BLOCKED: {exc}", file=sys.stderr)
        return 1
    print(f"GATE PASSED: {datasource} ingested ({task_id}), sized, and monitored")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Three properties make this gate safe to wire into a real pipeline. Fail-fast ordering: the offline stage-1 check runs before any Overlord slot is consumed, so a malformed spec costs milliseconds, not an ingestion window. Handoff, not task success: stage 2 only returns after the Coordinator serverview confirms queryability, closing the "passed but not there" gap. Capped backoff everywhere: every cluster call retries with exponential backoff to a fixed ceiling, so a leader election or transient partition during the deploy window does not produce a spurious block. The loadstatus shape in stage 4 is simplified for illustration; in a real deploy, wire the alert-manager check alongside it so a firing lifecycle alert also blocks.

Failure Modes & Diagnostics #

When the gate blocks, the stage name in the exit message tells you which surface to inspect. Reproduce each stage's decision by hand from the same REST APIs the gate uses.

1. Stage 1 blocks — spec rejected before submission. The offline contract found a structural or semantic problem. Reproduce the required-key and pairing checks against the spec file:

jq '{type, spec_keys: (.spec | keys),
     forceGR: .spec.tuningConfig.forceGuaranteedRollup,
     partType: .spec.tuningConfig.partitionsSpec.type,
     intervals: .spec.dataSchema.granularitySpec.intervals}' spec.json

A forceGR: true with partType: "dynamic", or a null intervals, is exactly what stage 1 rejects.

2. Stage 2 blocks — task FAILED. Pull the full report and read the error message before re-running:

curl -s "http://overlord:8090/druid/indexer/v1/task/$TASK_ID/reports" \
  | jq '.ingestionStatsAndErrors.payload.errorMsg'

3. Stage 2 blocks — succeeded but never queryable. The task reached SUCCESS but the serverview stayed empty. Check whether the Coordinator has assigned the segments and whether the target tier has headroom:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadstatus?full" | jq
curl -s "http://coordinator:8081/druid/coordinator/v1/loadqueue?simple" | jq

A non-empty load queue with no capacity means the destination tier is full under druid.server.maxSize.

4. Stage 3 blocks — size regression. Read back the produced distribution the same way the audit does:

curl -s "http://coordinator:8081/druid/coordinator/v1/datasources/events_web/segments?full=true" \
  | jq '[.[] | (.size/1048576 | floor)] | {min: min, max: max, count: length}'

A spread dominated by sub-300 MB entries means targetRowsPerSegment is too low or the rollup ratio was mis-measured; a few oversized entries mean a skewed partitionDimensions.

5. Stage 4 blocks — under-replicated or unavailable. Confirm the counts the gate read:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadstatus?full" \
  | jq 'to_entries[] | select(.value != {}) | {datasource: .key, pending: .value}'

Persistent nonzero counts after the load queue drains point at insufficient tier capacity or a mid-restart Historical.

Automation Checklist #

Wire these gates into the pipeline so every ingestion deploy runs the same four stages and no stage can be skipped under time pressure.

• Offline spec validation first — run the structural and semantic checks against the assembled JSON before any Overlord call, so a malformed spec never consumes a slot.
• Required-key assertion — confirm type, spec.dataSchema, spec.ioConfig, and spec.tuningConfig are all present and non-empty.
• Rejected-pairing check — fail locally on forceGuaranteedRollup + dynamic partitioning, or dropExisting without bounded intervals.
• Submit with capped backoff — POST to the Overlord and poll status to a terminal state, retrying transient failures with exponential backoff to a fixed ceiling.
• Gate on handoff, not task success — poll the Coordinator serverview until the interval's segments are queryable before advancing.
• Post-ingest size audit — read back the produced segment distribution and block when the out-of-band fraction exceeds tolerance.
• Compaction follow-up on drift — when the distribution has drifted, trigger a compaction pass and re-audit rather than shipping the regression.
• Replication check — assert no segments are unavailable or under-replicated for the datasource before exiting zero.
• Alert-clean check — verify no lifecycle alert is firing for the affected datasource as the final stage-4 condition.
• Single non-zero exit contract — every failure path returns a non-zero code naming the stage and reason so CI logs point straight at the fix.

Related #

• Schema validation for Druid specs — the pre-flight contract that powers stage 1, rejecting malformed or incompatible specs before the Overlord sees them.
• Dynamic ingestion spec generation — how the spec this gate validates is assembled at runtime from catalog metadata and cluster telemetry.
• Ingestion handoff validation — the queryability confirmation that stage 2 depends on to gate on handoff rather than task success.
• Automated compaction task scheduling — the compaction mechanism stage 3 triggers when the segment-size audit detects drift.
• Optimizing segment size for Historical nodes — the target size band the stage-3 audit holds every deploy accountable to.
• Blocking deploys on segment size regression — the stage-3 pass/fail decision worked through end to end.

Up one level: Apache Druid segment architecture & lifecycle fundamentals frames how validation, ingestion, compaction, and monitoring compose into one governed segment lifecycle.