Idempotent Ingestion & Retry Patterns for Apache Druid

Any automated pipeline that submits work to Apache Druid will, sooner or later, submit the same work twice — a scheduler retries a timed-out task, a network partition hides a successful response, an operator re-runs a backfill they think failed. The difference between a pipeline that tolerates this and one that silently double-counts revenue is whether ingestion is idempotent: whether re-running identical work leaves the datasource in the same state as running it once. This guide, part of automated ingestion pipeline orchestration, covers the four mechanisms that make Druid ingestion safe to retry — deterministic content-hashed task ids, the Overlord's duplicate-id rejection, dropExisting with bounded intervals for atomic replacement, and capped exponential backoff — and it draws the line between the at-least-once delivery an orchestrator provides and the exactly-once effect those mechanisms deliver on top of it.

Mechanics & Internals #

Idempotency in Druid rests on one structural fact: every segment carries a version, and a query only ever reads the single newest version that covers each interval. The segment identifier is datasource_interval_version_partitionNum, where the version is an ISO timestamp assigned at publish time. When two segments cover the same interval, the Broker's timeline resolves the overlap in favor of the higher version and ignores the older one entirely — it is overshadowed, not deleted, and stops being served the moment the newer version loads. This is the primitive that makes safe re-ingestion possible: a corrected re-run does not mutate the old data, it publishes a new version that atomically wins. Everything in this guide is about making sure a retry produces either no new work at all, or a clean new version that fully supersedes the old — never a partial overlap that leaves two versions live and double-counts rows.

The first mechanism is the deterministic task id. Druid lets the caller supply a task id on submission via the spec's top-level id field; if omitted, the Overlord generates a random one. A random id is the root cause of most double-ingestion incidents, because an orchestrator retry that mints a fresh id submits work the Overlord sees as brand new. Deriving the id from a content hash of the datasource, the target interval, and a spec version instead means every retry of the same logical work computes the same id. The hash must include the spec version so that a genuinely changed spec — new dimensions, different rollup — produces a new id and is allowed to run; two runs collide only when they are truly identical work. This is the same content-hashing the parent guide and dynamic ingestion spec generation use, elevated here to the load-bearing idempotency key.

The second mechanism is Overlord duplicate rejection. When a task with an id that already exists in the task metadata store is submitted, the Overlord refuses it rather than running a second copy. The response is an HTTP 400 whose body names the conflicting id. The critical implementation detail is that the orchestrator must treat this specific rejection as success, not as an error to retry — the work is already done or in flight, which is exactly the outcome the retry wanted. Conflating a duplicate-id 400 with a validation 400 (a malformed spec, which is genuinely fatal) is a common bug; the two are distinguished by inspecting the message for the submitted id. Together, the deterministic id and this rejection convert an at-least-once delivery channel into an at-most-once execution, and because the work either ran or is running, the combined effect is exactly-once.

The third mechanism is dropExisting with bounded intervals, which handles the case where re-ingestion should run because the data changed. Setting ioConfig.dropExisting: true on an index_parallel task with an explicit granularitySpec.intervals tells Druid that, on handoff, the new segments atomically replace every existing segment within that exact interval — including any segments the new run did not itself produce. This is what makes a backfill safe: without it, a re-run with appendToExisting: true adds a second version on top of the first and both stay live, double-counting; with it, the interval is cleanly swept and only the new version survives. The interval must bound the window precisely, because dropExisting will drop adjacent data if the interval is wider than intended — the atomic-replace guarantee is exactly as safe as the interval is tight. The dedicated walkthrough of this pattern lives in exactly-once Druid backfills.

The fourth mechanism is capped exponential backoff with a retry budget, which governs how a pipeline retries transient failures without becoming a denial-of-service against its own Overlord. Not every failure should be retried: a 400 validation error is deterministic and will fail identically forever, so retrying it wastes the SLA, while a 503 or a connection reset is transient and worth retrying. The pipeline must also cap total attempts and eventually route a persistently failing task to a dead-letter path — a poison task that fails every time is often a spec bug or a corrupt input, and retrying it indefinitely starves healthy work. The interplay of these four mechanisms with the run-phase polling loop is the subject of async task execution patterns, and the whole retry contract is what lets an external scheduler drive Druid safely, as covered in workflow orchestrator integration.

Validated Configuration Spec #

The spec an idempotent retry submits is an ordinary index_parallel batch spec with two load-bearing choices: a pinned id and dropExisting: true bounded to an exact interval. Every required top-level key is present and the block is copy-ready against the latest stable Druid release.

{
  "type": "index_parallel",
  "id": "idx_events_web_9f3c2a1b7e40",
  "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,
      "intermediatePersistPeriod": "PT10M",
      "forceGuaranteedRollup": true,
      "partitionsSpec": {
        "type": "range",
        "partitionDimensions": ["country"],
        "targetRowsPerSegment": 5000000
      }
    }
  }
}

The two idempotency-critical fields are the top-level id — pinned to a content hash so a resubmission collides and is rejected — and ioConfig.dropExisting: true, which, paired with the tightly bounded granularitySpec.intervals, guarantees that whichever run does execute replaces the interval atomically. appendToExisting is explicitly false: append plus a fresh id is the exact combination that double-counts, so an idempotent pipeline never uses it for backfills. The forceGuaranteedRollup: true setting requires the non-dynamic range partitioning and bounded interval shown, both of which the retry supplies by construction.

Sizing Heuristics & Formulas #

Retry policy is a sizing problem: too aggressive and the pipeline hammers a struggling Overlord, too timid and a transient blip becomes a missed SLA. With an initial delay ( d_0 ) doubling to a ceiling ( d_{max} ), the delay before attempt ( n ) is ( \min(d_0 \times 2^{,n-1},, d_{max}) ), and the worst-case total time spent waiting across ( R ) retries — the budget the orchestrator must allow before declaring the task dead — is:

$$ T_{retry} \approx \sum_{n=1}^{R} \min!\left(d_0 \times 2^{,n-1},\ d_{max}\right) $$

For ( d_0 = 2\text{s} ), ( d_{max} = 30\text{s} ), and ( R = 6 ), the delays are 2, 4, 8, 16, 30, 30 seconds, so ( T_{retry} \approx 90\text{s} ) of pure backoff on top of the task runtimes themselves. Add full jitter — sample each delay uniformly from ( [0, \min(d_0 \times 2^{,n-1}, d_{max})] ) — so that many pipelines retrying after a shared Overlord outage do not synchronize into a thundering herd; jitter halves the expected simultaneous load without materially lengthening the budget.

The second calibration is the poison-task threshold: how many consecutive failures of the same deterministic id prove the failure is structural rather than transient. If a task fails ( R ) times with the same id and the same error class, the probability it is a transient fault rather than a deterministic bug falls off geometrically, so a threshold of ( R = 3 ) to ( 5 ) identical-error failures is a reasonable cut-off before dead-lettering:

$$ \text{poison} \iff \text{failCount}(id) \geq R_{poison} \ \wedge\ \text{errorClass constant} $$

Finally, size the retry concurrency so a burst of retries after a recovered outage does not itself overwhelm capacity. If ( P ) pipelines all retry at once and each consumes ( k+1 ) worker slots, the recovery surge demands ( P \times (k+1) ) slots; stagger retries with jitter so the surge spreads across several ( d_{max} ) windows rather than landing in one. These bounds tie directly into the segment-size targets of segment size optimization strategies, because a replaced interval must still land segments in the target band, not fragment it.

Python Orchestration Snippet #

The reference retry wrapper below uses only the standard library plus requests. It classifies failures into retryable and fatal, applies capped exponential backoff with full jitter, treats a duplicate-id rejection as success, and dead-letters a poison task after a bounded number of identical failures.

import time
import random
import hashlib
import requests

OVERLORD = "http://overlord:8090"
RETRYABLE = {500, 502, 503, 504}


def deterministic_task_id(datasource: str, interval: str, spec_version: str) -> str:
    """Same logical work -> same id; a changed spec_version -> a new id."""
    key = f"{datasource}:{interval}:{spec_version}".encode()
    return f"idx_{datasource}_{hashlib.sha1(key).hexdigest()[:12]}"


def submit_idempotent(spec: dict, task_id: str, max_attempts: int = 6) -> str:
    """Submit with a pinned id. Duplicate-id 400 is success; 4xx is fatal."""
    spec = {**spec, "id": task_id}
    delay = 2.0
    for attempt in range(max_attempts):
        try:
            r = requests.post(f"{OVERLORD}/druid/indexer/v1/task",
                              json=spec, timeout=30)
            if r.status_code == 400 and task_id in r.text:
                return task_id                      # already ran: idempotent win
            if 400 <= r.status_code < 500:
                r.raise_for_status()                # fatal: malformed spec, no retry
            if r.status_code in RETRYABLE:
                raise requests.HTTPError(f"transient {r.status_code}")
            r.raise_for_status()
            return r.json()["task"]
        except (requests.ConnectionError, requests.Timeout, requests.HTTPError) as exc:
            if attempt == max_attempts - 1:
                raise
            sleep = random.uniform(0, min(delay, 30.0))   # full jitter, capped
            time.sleep(sleep)
            delay = min(delay * 2, 30.0)
    raise RuntimeError("unreachable")


def run_with_poison_guard(spec: dict, datasource: str, interval: str,
                          spec_version: str, poison_after: int = 3) -> str:
    """Bounded retries of the SAME id; dead-letter a structurally failing task."""
    task_id = deterministic_task_id(datasource, interval, spec_version)
    last_state, fails = None, 0
    while fails < poison_after:
        submit_idempotent(spec, task_id)
        state = poll_terminal(task_id)
        if state == "SUCCESS":
            return task_id
        last_state, fails = state, fails + 1
    raise PoisonTaskError(f"{task_id} failed {fails}x (last={last_state}); dead-lettered")


def poll_terminal(task_id: str, timeout_s: int = 3600) -> str:
    deadline = time.monotonic() + timeout_s
    delay = 5.0
    while time.monotonic() < deadline:
        r = requests.get(f"{OVERLORD}/druid/indexer/v1/task/{task_id}/status",
                         timeout=15)
        r.raise_for_status()
        state = r.json()["status"]["status"]
        if state in {"SUCCESS", "FAILED"}:
            return state
        time.sleep(delay)
        delay = min(delay * 2, 30.0)
    raise TimeoutError(f"{task_id} not terminal within {timeout_s}s")


class PoisonTaskError(RuntimeError):
    """Raised when a deterministic task fails structurally and must be dead-lettered."""

Three properties make this safe. Failure classification: only 5xx, connection, and timeout errors are retried; a 4xx validation failure fails fast rather than burning the retry budget on a spec that will never parse. Full jitter: the sleep is sampled from [0, capped delay], so a fleet recovering from a shared outage spreads its resubmissions instead of synchronizing. Poison-task dead-lettering: the same deterministic id is retried only a bounded number of times, after which PoisonTaskError routes the task to human triage rather than looping forever — the exact structural failures diagnosed in debugging Druid supervisor task failures.

Failure Modes & Diagnostics #

Idempotency failures are quiet — the pipeline reports success while the data is wrong — so diagnosis leans on inspecting segment versions directly.

1. Duplicate rows after a retry. The classic symptom of a non-deterministic id plus appendToExisting. Check how many live versions cover the interval:

curl -s "http://coordinator:8081/druid/coordinator/v1/datasources/events_web/segments?full=true" \
  | jq 'group_by(.interval)[] | {interval: .[0].interval, versions: (map(.version) | unique)}'

More than one version for an interval that should have been replaced means the retry appended instead of superseding — pin the id and switch to dropExisting: true.

2. A retry ran when it should have been a no-op. The content hash changed even though the work was identical — usually because the hash included a volatile field like a timestamp or a random nonce. Recompute the id offline and confirm two identical runs collide:

python3 -c "import hashlib; k=b'events_web:2026-07-16T00:00:00Z/2026-07-17T00:00:00Z:v3'; \
  print('idx_events_web_'+hashlib.sha1(k).hexdigest()[:12])"

Both runs must print the same id. If they differ, remove the volatile input from the hash key.

3. A poison task retries forever. A structurally broken spec fails every attempt but the wrapper keeps resubmitting. Pull the error message and confirm it is deterministic, not transient:

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

A constant parse or class-cast error across attempts is a poison task — dead-letter it and fix the spec.

4. A valid retry is rejected as a duplicate but the original actually failed. The prior task with that id ended FAILED, and Druid still considers the id used. Confirm the prior task's terminal state, then resubmit under a bumped spec_version so a fresh id is minted:

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

5. dropExisting dropped adjacent data. The interval was wider than the data being replaced, so the atomic sweep removed neighboring segments. Verify the replaced interval matches the intended window exactly by comparing the spec's intervals against the loaded segment boundaries — the failure and its guard rails are worked through in exactly-once Druid backfills.

Automation Checklist #

Wire these into any pipeline that may resubmit work to Druid.

• Content-hashed task id — derive the top-level id from datasource, interval, and spec version; never let the Overlord mint a random id for retryable work.
• Include spec version in the hash — a changed spec must produce a new id so it is allowed to run; identical work must collide.
• Exclude volatile fields from the hash — no timestamps, nonces, or run counters in the id key, or retries stop colliding.
• Treat duplicate-id 400 as success — inspect the message for the submitted id and return success rather than retrying.
• Classify failures before retrying — retry only 5xx/connection/timeout; fail fast on 4xx validation errors.
• Cap backoff and add full jitter — start small, ceil at 30s, sample the actual delay to avoid a thundering herd on recovery.
• Bound retries with a poison threshold — dead-letter a deterministic id after a fixed number of identical failures.
• Use dropExisting: true for backfills — never appendToExisting for re-ingestion; pair dropExisting with an exact intervals.
• Assert interval tightness — confirm the spec's intervals matches the window being replaced so dropExisting cannot touch neighbors.
• Audit live versions post-handoff — confirm exactly one version covers each replaced interval before signalling success.

Related #

• Exactly-once Druid backfills — replacing an interval atomically with dropExisting: true and verifying only the newest version is served.
• Workflow orchestrator integration — driving these idempotent submissions from Airflow or Dagster with safe retries and SLAs.
• Ingestion handoff validation — confirming a retried task's segments actually loaded before marking the run done.
• Dynamic ingestion spec generation — the spec builder whose version participates in the idempotency hash.
• Async task execution patterns — the submit-poll-backoff loop these retry policies extend.

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