Apache Druid Segment Replication, Balancing & High Availability: Production Operations Guide

Apache Druid stays queryable through node failures because the Coordinator continuously reconciles a declarative target — how many replicas of each segment should sit on which Historical tier — against the live cluster, re-replicating and rebalancing until reality matches intent. For OLAP data engineers, platform developers, and DevOps teams, this control loop is the single mechanism that decides whether a lost Historical is a non-event or a paging incident, and it is governed almost entirely by load rules, a replication factor per tier, and a balancing strategy. This guide dissects that loop end to end: the placement mechanics, the exact configuration surface, the capacity math, the orchestration patterns that safely nudge it, the failure signatures it produces, the trust boundaries it crosses, and the metrics that tell you it is healthy.

Architecture at a Glance #

The diagram traces one segment's replica path: the Coordinator reads the metadata store and a datasource's load rules, decides a per-tier replica count, enqueues loads onto Historicals across a hot and a cold tier, and the Broker then routes each incoming query to whichever replica is available — surviving the loss of any single Historical that still leaves one replica standing.

Every arrow is a place where intent and reality can diverge: a rule can demand two hot replicas the tier lacks capacity to hold, a balancing pass can move segments faster than Historicals can pull them, or a dead node can leave a segment under-replicated until the Coordinator notices and re-copies it. The sections below treat each as an operational surface with its own tunables, sizing math, failure signatures, and metrics. This guide sits alongside the segment architecture and lifecycle fundamentals — that guide defines what a segment is; this one defines how many copies of it live where, and how the Druid cluster keeps that true.

Core Concept & Internal Mechanics #

Placement in Druid is a reconciliation loop, not a one-time assignment. Once per Coordinator run (governed by druid.coordinator.period, default PT60S), the Coordinator runs a fixed sequence of duties: it reads every used segment's descriptor from the metadata store, evaluates the datasource's ordered load rules to compute the desired replica count per tier, compares that against what each Historical currently advertises, and then enqueues the differences — loads for under-replicated segments, drops for over-replicated or rule-excluded ones, and moves to even out utilization. Because the loop is declarative, the Druid cluster is self-healing: you describe the target state in rules once, and every subsequent run drives toward it regardless of what failed in between.

Three concepts do the actual work:

• Load rules are an ordered list attached to each datasource (with a Druid cluster-wide _default fallback). Each rule matches a set of segments by interval (loadByInterval), by rolling age (loadByPeriod), or unconditionally (loadForever), and carries a tieredReplicants map of tier name to replica count. The Coordinator walks the list top-down and applies the first matching rule to each segment — so rule order is load-bearing, and a loadForever catch-all must sit last. The full mechanics of matching and tier assignment are the subject of Coordinator load rules and tier placement.
• Replication factor is simply the sum of a segment's tieredReplicants values across all tiers. A segment loaded as {"hot": 2, "cold": 1} has a replication factor of three and tolerates two simultaneous replica losses before becoming unavailable. This factor — and the difference between under-replicated (fewer copies than desired, but at least one, so still queryable) and unavailable (zero loaded copies) — is the heart of segment replication and fault tolerance.
• Tiers partition the Historical fleet into named pools via the runtime property druid.server.tier on each Historical. A tier is the unit that a tieredReplicants entry addresses, letting you keep recent data triple-replicated on fast NVMe nodes and archival data single-replicated on cheap high-density nodes. Segments never span tiers by accident; they land only where a rule sends them.

Once the desired counts are known, the balancing strategy decides which specific Historical within a tier receives each new replica and which segments to relocate to level the fleet. The default cost strategy scores every candidate placement with a cost function that penalizes putting segments from the same datasource and adjacent time intervals on the same server, which naturally spreads a datasource's segments — and therefore query load — across the tier. The trade-offs between cost, cachingCost, and random, and the knobs that keep balancing from thrashing, are the domain of Historical tier balancing strategies.

The placement reconciliation loop #

A segment's replicas are never simply "placed." They occupy a count that the Coordinator continuously drives toward the rule-declared target:

1. Desired — the load rules resolve to a per-tier replica count for the segment. This is intent, computed fresh every run.
2. Enqueued — for each tier short of its desired count, the Coordinator picks target Historicals via the balancing strategy and appends load commands to their load queues, throttled by replicationThrottleLimit so a backlog never floods the fleet.
3. Loaded / Available — the Historicals pull the bytes from deep storage, memory-map them, and announce the segment; the Broker adds each replica to its timeline. Only now is the replica serving queries.
4. Balanced — subsequent runs may move already-loaded replicas between Historicals in the same tier to even out disk and query load, dropping the source copy only after the destination confirms load.

Because Historicals hold only copies — the durable segment always lives in deep storage — the loop can recover from any node loss by re-reading bytes that were never lost. That is the structural reason a Druid Historical is disposable: it is a cache, not a system of record.

Configuration Reference #

Placement behaviour is split across three configuration surfaces: load rules (per datasource, via the rules API), the Coordinator dynamic config (cluster-wide, hot-reloadable via API), and a few Historical runtime properties (per node, set at boot). The rule document below is posted to POST /druid/coordinator/v1/rules/{dataSource} and is evaluated top-down; the first matching rule wins for each segment.

[
  {
    "type": "loadByPeriod",
    "period": "P7D",
    "includeFuture": true,
    "tieredReplicants": { "hot": 2, "cold": 1 }
  },
  {
    "type": "loadByPeriod",
    "period": "P3M",
    "includeFuture": false,
    "tieredReplicants": { "cold": 1 }
  },
  {
    "type": "dropForever"
  }
]

Read top to bottom: segments from the last 7 days (and any future-dated ones) get two hot replicas plus one cold; older segments back to 3 months get a single cold replica; everything older is dropped from Historicals (its bytes remain in deep storage). The concrete hot/cold recipes and the X-Druid-Author/X-Druid-Comment audit headers are worked through in writing tier-aware load rules.

The Coordinator dynamic config governs how aggressively the loop acts. It is read and written as JSON at GET/POST /druid/coordinator/v1/config:

{
  "maxSegmentsToMove": 100,
  "replicationThrottleLimit": 500,
  "replicantLifetime": 15,
  "balancerComputeThreads": 2,
  "maxSegmentsInNodeLoadingQueue": 500,
  "useRoundRobinSegmentAssignment": true,
  "smartSegmentLoading": true,
  "decommissioningNodes": [],
  "decommissioningMaxPercentOfMaxSegmentsToMove": 70,
  "pauseCoordination": false
}

The fields that materially shape replication and balancing:

• smartSegmentLoading — when true (the default in current Druid), the Coordinator auto-computes maxSegmentsToMove, replicationThrottleLimit, maxSegmentsInNodeLoadingQueue, and useRoundRobinSegmentAssignment from cluster size and forces the cost balancer. Set it false only when you must pin these values manually; any explicit values you set are ignored while it is on.
• replicationThrottleLimit — the maximum number of segment replicas (beyond the first copy) the Coordinator will enqueue to a tier in a single run. This is the primary brake that stops a large under-replication event from saturating deep-storage read bandwidth.
• maxSegmentsToMove — the cap on balancing moves per run. Higher converges utilization faster but risks a move storm; lower is calmer but slower to level.
• replicantLifetime — how many Coordinator runs a replica may sit un-loaded in a queue before the Coordinator raises an alert about it, the signal that a tier cannot keep up.
• maxSegmentsInNodeLoadingQueue — the depth cap on any one Historical's load queue, preventing a single slow node from accumulating unbounded work.
• useRoundRobinSegmentAssignment — assign new segments round-robin (cheap) and let the balancer relocate them later, instead of running the full cost computation for every initial assignment; a large-cluster speedup.
• decommissioningNodes / decommissioningMaxPercentOfMaxSegmentsToMove — mark Historicals for graceful drain and bound how much of the move budget the drain may consume, so decommissioning never starves normal balancing.

Finally, three Historical runtime properties complete the picture, set in each node's runtime.properties:

• druid.server.tier — the tier name this Historical joins (default _default_tier). This is what a tieredReplicants key must match exactly.
• druid.server.priority — an integer tie-breaker; when a segment could be served from multiple tiers, the Broker prefers the higher-priority tier. Give hot tiers a higher priority than cold.
• druid.server.maxSize — the byte capacity the Historical advertises for segments, the denominator in every utilization and headroom calculation below.

Operational Sizing & Constraints #

Replication multiplies storage. The total Historical capacity a datasource consumes is its deep-storage footprint times its replication factor, summed per tier — and the fleet must hold that with headroom for the balancer to work:

$$\text{tierCapacityNeeded} \approx \sum_{\text{intervals in tier}} \text{segmentBytes} \times \text{replicas}_{\text{tier}}$$

For a datasource whose last 7 days occupy 900 GB at two hot replicas, the hot tier must hold roughly $900 \times 2 = 1800$ GB of that datasource alone, before any other datasource or headroom. Provision each tier so steady-state utilization stays below about 85% of aggregate druid.server.maxSize:

$$\text{utilization} = \frac{\sum \text{loadedBytes}}{\sum \text{maxSize}} \lesssim 0.85$$

The remaining headroom is not slack — it is what lets the Coordinator move segments and, critically, re-replicate after a node dies. A tier running at 95% cannot absorb a lost node's replicas, so an under-replication event there becomes a sustained unavailability event. The rule of thumb for surviving the loss of $f$ Historicals in a tier of $N$ nodes each of size $s$ is that the survivors must still fit the whole working set:

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

So a tier that must tolerate one node loss ($f = 1$) needs at least one node's worth of spare capacity distributed across the survivors. Replication factor and this capacity constraint are two views of the same budget: a replica you cannot place is not a replica. The band each individual segment should occupy so a Historical can memory-map it without heap pressure — roughly 300–700 MB — is derived in optimizing segment size for Historical nodes; replication does not change that band, it multiplies how many copies of it the fleet stores.

A second constraint bounds convergence speed. Re-replicating $M$ segments averaging $b$ bytes, throttled to $r$ replicas per run over a period $p$, and bounded by aggregate deep-storage read throughput $T$, takes at least:

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

Whichever term dominates tells you the true bottleneck: raise replicationThrottleLimit when the first term wins, add deep-storage read bandwidth when the second does.

Pipeline Orchestration Patterns #

Replication and balancing are mostly autonomous, but production still needs to drive them deliberately: push a new rule set, wait until the Druid cluster has actually converged before declaring a migration done, or throttle moves during a maintenance window. The pattern is always the same — mutate the declarative target, then poll an authoritative status endpoint to a converged state with capped exponential backoff, never assuming the change took effect instantly.

The stdlib + requests orchestrator below updates a datasource's load rules, then blocks until the Coordinator reports zero under-replicated and zero unavailable segments for it, so a downstream step (a tier migration, a decommission) only proceeds once replicas are genuinely in place:

import time
import requests

COORDINATOR = "http://druid-coordinator:8081"


def set_rules(datasource: str, rules: list, author: str = "orchestrator") -> None:
    """Replace the load rules for a datasource (audited)."""
    r = requests.post(
        f"{COORDINATOR}/druid/coordinator/v1/rules/{datasource}",
        json=rules,
        headers={
            "Content-Type": "application/json",
            "X-Druid-Author": author,
            "X-Druid-Comment": "automated rule update",
        },
        timeout=30,
    )
    r.raise_for_status()


def load_status(datasource: str) -> dict:
    """Per-tier counts still to load for the datasource."""
    r = requests.get(
        f"{COORDINATOR}/druid/coordinator/v1/loadstatus",
        params={"full": "", "forceMetadataRefresh": "false"},
        timeout=30,
    )
    r.raise_for_status()
    return r.json().get(datasource, {})


def wait_for_convergence(datasource: str, max_wait: float = 1800.0) -> bool:
    """Block until nothing remains to load for the datasource across all tiers."""
    delay, waited = 5.0, 0.0
    while waited < max_wait:
        remaining = load_status(datasource)
        if not remaining or all(v == 0 for v in remaining.values()):
            return True
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, 60.0)  # exponential backoff, capped
    return False


def apply_and_verify(datasource: str, rules: list) -> None:
    set_rules(datasource, rules)
    if not wait_for_convergence(datasource):
        raise RuntimeError(f"{datasource}: replicas did not converge within budget")
    print(f"{datasource}: rules applied and fully replicated")

The loadstatus?full endpoint returns, per datasource and tier, the number of segments still queued to load — the authoritative "are we there yet." The same shape generalizes to every placement action: throttling balancing before a deploy is a POST /druid/coordinator/v1/config that lowers maxSegmentsToMove, then a poll of segment/loadQueue/count back to baseline. This backoff-and-verify discipline mirrors the ingestion side covered in dynamic ingestion spec generation; for cold-to-warm data movement specifically, the tiered storage migration runbook drives exactly this rule-then-converge cycle across tiers.

Failure Modes & Diagnostics #

Each failure is stated as symptom → root cause → remediation, in rough order of how often it bites production clusters.

1. Segments show as unavailable; queries return partial or empty results. → Zero loaded replicas: either the destination tier is at capacity, or a rule references a tier name no Historical advertises. → Check GET /druid/coordinator/v1/loadstatus?full and GET /druid/coordinator/v1/tiers; confirm every tieredReplicants key matches an actual druid.server.tier, and free capacity by dropping older intervals or adding nodes.
2. under-replicated count stuck above zero for many runs. → The tier lacks capacity for the full replica count, or replicationThrottleLimit is too low to catch up after a large event. → Add capacity or lower the replica count; if throughput is the limit, raise the throttle (or disable smartSegmentLoading to pin it) and confirm deep-storage read bandwidth.
3. A dead Historical's segments never re-replicate. → The surviving nodes in the tier have no headroom, so the Coordinator has nowhere to place the copies. → This is the capacity-headroom constraint biting; the full recovery procedure is in recovering from Historical node loss.
4. Coordinator run time climbing; segments constantly moving ("balancing storm"). → maxSegmentsToMove too high for the Druid cluster, a flapping node, or many tiny segments inflating the cost computation. → Lower maxSegmentsToMove, stabilize the flapping node, and compact tiny segments; the full triage is in diagnosing segment balancing storms.
5. New rule set appears to have no effect. → Rule order: an earlier loadForever or broad loadByPeriod matched first and shadowed the new rule, or the change hit the wrong datasource instead of _default. → Fetch GET /druid/coordinator/v1/rules/{ds} and read top-down; the first match wins, so move specific rules above general ones.
6. Query latency spikes on one Historical while others idle. → Skewed placement — the balancer put many hot-datasource segments on one node, or a recently added node hasn't received its share yet. → Confirm the cost strategy is active and let balancing run; if it persists, the strategy or maxSegmentsToMove is mis-tuned per the balancing strategies guide.

Security & Access Control Boundaries #

Placement crosses the same three trust planes as the rest of the segment lifecycle, and each must be locked independently.

• Coordinator control plane. The rules and config APIs (/druid/coordinator/v1/rules/*, /druid/coordinator/v1/config) are what decide where data lives and how many copies exist. A principal who can WRITE these can silently drop a datasource from all Historicals — a denial-of-service on your own data — or force a replica onto a tier that violates a data-residency policy. Gate these endpoints behind the authorization extension with CONFIG/STATE write permission restricted to operators, and rely on the X-Druid-Author/X-Druid-Comment audit trail to attribute every rule change.
• Deep storage plane. Historicals re-replicate by reading raw bytes from deep storage, so every serving node's service identity needs read access to the segment prefix. That access is Druid-opaque: anyone with those object-store credentials reads the columnar bytes directly, bypassing datasource ACLs. Scope the bucket policy so only Historical and indexing identities can read the prefix, with encryption at rest enabled.
• Metadata plane. The Coordinator reasons entirely over segment descriptors and the used flag in the metadata store; write access there lets an attacker mark segments unused (dropping them from serving) or forge placement intent. Restrict metadata connections to Druid service accounts only.

Tiering also encodes a coarse access boundary: an isolated tier (for example a regulated-data tier reachable only from a restricted Broker) plus a rule that pins those segments to it keeps sensitive data physically off general-purpose nodes. The complete hardening model across all three planes is detailed in security boundaries for segment access, and the rule syntax that expresses per-tenant placement and expiry in configuring segment retention policies.

Monitoring & Alerting Hooks #

The Coordinator emits placement metrics every run; scraped into Prometheus (via the prometheus-emitter extension or a StatsD bridge) they give direct visibility into the reconciliation loop. The table maps the highest-value signals to their meaning and a starting alert threshold.

Practical panel layout for a Grafana placement dashboard: one availability row (segment_unavailable_count and segment_underreplicated_count as stat panels that go red on nonzero — the earliest and clearest signal that replication intent isn't met), one flow row (per-server loadqueue_count/size and segment_moved_count time series to see whether queues drain and whether balancing is calm or storming), one capacity row (segment_used_percent per Historical as a heatmap, with the 85% line drawn), and one control-plane row (coordinator_time and tier_historical_count). Alert on segment_unavailable_count > 0 as page-worthy — it is the unambiguous statement that a query cannot see data it should. Wire these alerts to the same orchestrator that pushes rules, so a failed convergence check and a firing alert tell one coherent story, exactly as the lifecycle guide's monitoring hooks recommend for the ingestion side.

Related #

• Coordinator load rules and tier placement — how ordered load rules resolve a per-tier replica count and pin segments to Historical tiers.
• Segment replication and fault tolerance — replication factor, under-replicated vs unavailable counts, throttling, and the availability guarantees they buy.
• Historical tier balancing strategies — choosing cost, cachingCost, or random and tuning the move budget for even utilization.
• Query routing and segment discovery — how the Broker builds its timeline over the replicas this loop places and routes around a missing one.
• Automated compaction task scheduling — consolidating tiny segments so the balancer and Coordinator loop stay cheap.

Up one level: this guide is one of the site's foundations — return to the segment management home to explore the architecture, ingestion, and compaction guides alongside it.