Druid Segment Ops Segment & Ingestion Orchestration

Segment Replication and Fault Tolerance in Apache Druid

Replication factor is the single number that decides whether losing a Historical is invisible to queries or an outage — and Druid tracks two very different states, under-replicated (fewer copies than desired, but still queryable) and unavailable (zero copies loaded, data missing from results), that operators routinely confuse until one pages them. This guide, part of Apache Druid segment replication, balancing & high availability, makes the distinction precise: how the replica count declared by load rules becomes an availability guarantee, how the replication throttle governs recovery speed, and how to size a tier so the loss of a node re-heals instead of degrading service.

Under-replicated versus unavailable states after Historical node loss A segment with replication factor three across three Historicals is fully replicated and queryable. Losing one Historical drops it to two of three copies: it is under-replicated but still queryable, and the Coordinator re-replicates the missing copy from deep storage onto a survivor. Losing all copies leaves zero of three: the segment is unavailable and queries return partial results until a copy reloads from deep storage. Healthy · 3 of 3 Under-replicated · 2 of 3 Unavailable · 0 of 3 replica · Hist A replica · Hist B replica · Hist C queryable lose Hist A lost replica · Hist B replica · Hist C still queryable Coordinator re-replicates from deep storage lose B + C lost lost lost partial results reload from deep storage restores availability deep storage always holds the durable copy — Historicals are caches, never the source of truth

Mechanics & Internals

Every used segment has a desired replica count — the sum of its tieredReplicants across tiers, computed each Coordinator run from the datasource's load rules — and an actual loaded count, the number of Historicals currently announcing it. The relationship between them defines three states the Coordinator tracks and emits as metrics:

  • Fully replicated: actual equals desired. Nothing to do.
  • Under-replicated: 0 < actual < desired. At least one copy is loaded, so the segment is still queryable, but the Druid cluster has less redundancy than the rules demand. The Coordinator enqueues the shortfall as new loads. This is counted by segment/underReplicated/count.
  • Unavailable: actual == 0 while desired > 0. No Historical serves the segment; queries touching its interval return partial results (or error, depending on Broker settings). This is counted by segment/unavailable/count and is the page-worthy state.

The crucial operational truth is that these are different severities. Under-replication is a redundancy warning — you have lost margin, not data. Unavailability is data loss from the query plane's perspective, even though the bytes remain safe in deep storage. Alerting must treat them differently: unavailable > 0 pages immediately; underReplicated > 0 warns and escalates only if it persists past the time a healthy cluster needs to re-replicate.

Recovery is automatic because Historicals hold only copies. When a Historical stops announcing (it crashed, was drained, or lost network), the Coordinator's next run sees the affected segments drop below desired count and enqueues fresh loads onto other Historicals in the same tier, which pull the bytes straight from deep storage. No data is ever reconstructed or lost — it is re-copied from the durable source. The detailed play-by-play of that recovery, including timeouts and forcing it along, is the subject of recovering from Historical node loss.

Two throttles shape how fast recovery happens, both in the Coordinator dynamic config. replicationThrottleLimit caps the number of replica loads (copies beyond the first) enqueued per tier per run, so a mass under-replication event ramps rather than saturating deep-storage read bandwidth. replicantLifetime bounds how many runs a queued replica may wait before the Coordinator raises an alert — the signal that the tier cannot keep pace. With smartSegmentLoading on (the current default), both are auto-sized from cluster shape; pin them manually only when you have measured a specific bottleneck.

A subtle but critical placement invariant underpins all of this: the Coordinator never puts two replicas of the same segment on the same Historical. Replication only buys fault tolerance if the copies live on distinct failure domains, so a replica count of three requires at least three Historicals in the tier — asking for three replicas in a two-node tier leaves every such segment permanently under-replicated. This is where replication meets the placement decisions in Coordinator load rules and tier placement and the levelling done by Historical tier balancing strategies.

Validated Configuration Spec

Fault tolerance is configured in two places: the replica counts in the load rules (how much redundancy you want) and the Coordinator dynamic config (how fast the loop restores it). A rule set that gives recent data three replicas across two tiers, posted to POST /druid/coordinator/v1/rules/{dataSource}:

[
  {
    "type": "loadByPeriod",
    "period": "P30D",
    "includeFuture": true,
    "tieredReplicants": { "hot": 2, "cold": 1 }
  },
  {
    "type": "loadForever",
    "tieredReplicants": { "cold": 1 }
  }
]

This gives the trailing 30 days a replication factor of three (tolerating two simultaneous replica losses) and everything older a single cold replica. The matching Coordinator dynamic config, read and written at GET/POST /druid/coordinator/v1/config, governs recovery aggressiveness:

{
  "smartSegmentLoading": true,
  "replicationThrottleLimit": 500,
  "replicantLifetime": 15,
  "maxSegmentsInNodeLoadingQueue": 500,
  "maxSegmentsToMove": 100
}
  • replicationThrottleLimit — with smartSegmentLoading: true this is auto-computed; the value shown applies only if you set smartSegmentLoading: false to pin it. It is the ceiling on replica loads per tier per run, the throttle that keeps a large recovery from monopolizing deep-storage reads.
  • replicantLifetime — how many runs a replica may sit queued before the Coordinator alerts; keep it low enough that a stuck tier surfaces quickly.
  • maxSegmentsInNodeLoadingQueue — per-Historical queue depth cap, so recovery load spreads instead of piling on one node.

The loadForever on the second rule guarantees the rule set is exhaustive — every segment matches something — so no segment is ever left in a silent unloaded limbo. The authoritative field semantics are in the Apache Druid coordinator documentation.

Sizing Heuristics & Formulas

Availability is a capacity property, not just a replica-count property. A tier of $N$ Historicals each advertising maxSize $s$ survives the loss of $f$ nodes only if the survivors can still hold the entire working set with balancing headroom:

$$(N - f) \times s \times 0.85 ;\geq; \text{workingSetBytes}_{\text{tier}}$$

So a hot tier holding 3.2 TB of loaded segments on 2 TB nodes needs, to tolerate one loss, $(N - 1) \times 2000 \times 0.85 \geq 3200$, giving $N - 1 \geq 1.88$, so $N \geq 3$ — three hot nodes minimum, even though two could physically hold 3.2 TB. The extra node is the failure budget. Replica count and node count are linked: a replica count of $R$ demands at least $R$ nodes in the tier (distinct-host invariant), and surviving $f$ losses while keeping every segment queryable demands $R > f$:

$$R \geq f + 1 \quad\text{and}\quad N \geq R$$

Recovery time after a loss is bounded by the throttle and by deep-storage read throughput $T$. Re-replicating $M$ orphaned segments averaging $b$ bytes, at throttle $r$ per run of period $p$:

$$t_{\text{recover}} \approx \max!\left(\left\lceil \tfrac{M}{r} \right\rceil \times p,; \tfrac{M \times b}{T}\right)$$

For 20,000 orphaned segments at $r = 500$ and $p = 60\text{s}$, the throttle term is $40 \times 60 = 2400\text{s}$ (~40 min); if those segments average 500 MB and deep storage serves 4 GB/s, the throughput term is $20000 \times 0.5 / 4 \approx 2500\text{s}$ — here the two bottlenecks are balanced, so improving either alone barely helps. Measure both before tuning. This recovery window is exactly how long a single-node loss leaves the tier at reduced redundancy, so keep it short relative to your mean time between failures.

Python Orchestration Snippet

A health gate belongs in every deploy and maintenance pipeline: before draining a node or declaring an operation done, assert the Druid cluster is fully replicated and nothing is unavailable. The gate below polls the Coordinator's availability counters with capped exponential backoff and distinguishes the two severities, using only the standard library plus requests.

import time
import requests

COORDINATOR = "http://coordinator:8081"


def availability(datasource: str | None = None) -> tuple[int, int]:
    """Return (unavailable, under_replicated) segment counts."""
    ua = requests.get(
        f"{COORDINATOR}/druid/coordinator/v1/loadstatus",
        params={"simple": ""}, timeout=30,
    )
    ua.raise_for_status()
    unavailable = ua.json()
    ur = requests.get(
        f"{COORDINATOR}/druid/coordinator/v1/loadstatus",
        params={"full": ""}, timeout=30,
    )
    ur.raise_for_status()
    under = ur.json()
    if datasource:
        unavail = int(unavailable.get(datasource, 0))
        under_ds = sum(under.get(datasource, {}).values())
        return unavail, under_ds
    return (sum(int(v) for v in unavailable.values()),
            sum(sum(t.values()) for t in under.values()))


def gate(datasource: str | None = None, allow_under: bool = False,
         max_wait: float = 1200.0) -> None:
    """Block until unavailable==0 (always) and under-replicated==0 (unless allowed)."""
    delay, waited = 5.0, 0.0
    while waited < max_wait:
        unavailable, under = availability(datasource)
        if unavailable == 0 and (allow_under or under == 0):
            print(f"healthy: unavailable={unavailable} under_replicated={under}")
            return
        if unavailable > 0:
            print(f"UNAVAILABLE {unavailable} segments — data missing from queries")
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, 60.0)  # capped exponential backoff
    raise RuntimeError("cluster did not reach required availability within budget")


if __name__ == "__main__":
    # Refuse to drain a node while anything is unavailable or under-replicated.
    gate(datasource="events", allow_under=False)

The gate reads loadstatus?simple (unavailable segment counts per datasource) and loadstatus?full (per-tier still-to-load counts, the under-replication view) and treats them with different strictness: unavailability always blocks, while under-replication can be tolerated for a read-only operation via allow_under. This is the same submit-and-verify discipline the ingestion pipeline uses in async task execution patterns — never act on assumed state, poll the authoritative endpoint.

Failure Modes & Diagnostics

1. Segments permanently under-replicated despite spare disk. The replica count exceeds the number of Historicals in the tier, violating the distinct-host invariant. Count nodes per tier and compare to the max replica request:

curl -s "http://coordinator:8081/druid/coordinator/v1/tiers?simple" | jq
curl -s "http://coordinator:8081/druid/coordinator/v1/rules/events" \
  | jq '[.[].tieredReplicants // {} | to_entries[] | {tier: .key, replicas: .value}]'

If a tier shows 2 nodes but a rule asks for 3 replicas there, every matched segment stays one copy short forever — add a node or lower the count.

2. Unavailable count spikes and stays up after a node loss. The tier has no headroom to place the orphaned copies. Check per-Historical fill against maxSize:

curl -s "http://coordinator:8081/druid/coordinator/v1/servers?simple" \
  | jq '[.[] | {tier, host, usedPct: ((.currSize / .maxSize * 100) | floor)}]'

Survivors near 100% cannot absorb the load; this is the capacity-headroom failure covered in recovering from Historical node loss.

3. Under-replication clears far too slowly. The throttle is the bottleneck. Watch the per-server load queues drain, and check whether smartSegmentLoading is masking a manual limit:

curl -s "http://coordinator:8081/druid/coordinator/v1/loadqueue?simple" | jq
curl -s "http://coordinator:8081/druid/coordinator/v1/config" \
  | jq '{smartSegmentLoading, replicationThrottleLimit, replicantLifetime}'

If queues shrink steadily the ramp is healthy; if they are flat, capacity or deep-storage throughput — not the throttle — is the limit.

Automation Checklist

Wire these into deploy and maintenance pipelines so redundancy is verified, not assumed.

Up one level: Apache Druid segment replication, balancing & high availability frames replication, balancing, and recovery as one control loop.

Recovering From Historical Node Loss Read more

Back to Apache Druid Segment Replication, Balancing & High Availability