Preventing Batch and Streaming Segment Overlap

A backfill job re-ingests yesterday's data to correct a late-arriving dimension, and for the hours it overlaps a live Kafka supervisor, every query now returns roughly double the true count. The cause is two sets of used segments covering the same interval: the streaming segments the supervisor already handed off, and the batch segments the backfill just published. Druid serves both, and the rows are summed twice. This guide shows how to make batch and streaming ingestion share a timeline without overlap — using dropExisting on a bounded interval, skipOffsetFromLatest to keep the supervisor off the backfill's window, and handoff coordination so the switch is atomic. It is the concrete double-counting case under synchronizing batch and streaming ingestion.

Failure Modes & Diagnostics #

Overlap double-counting is diagnosable directly from segment metadata: the signature is more than one used version, or two distinct ingestion sources, covering the same interval. The streaming supervisor writes segments continuously as it reads Kafka; a batch index_parallel task without dropExisting and a bounded intervals publishes additional segments rather than replacing the streaming ones. Inspect the timeline for the suspect window:

# List used segments for the interval, grouped by interval, showing versions and count
curl -s "http://<coordinator-host>:8081/druid/coordinator/v1/datasources/<datasource>/segments?full=true" \
  | jq 'map(select(.interval | startswith("2025-11-13")))
        | group_by(.interval)[]
        | {interval: .[0].interval, versions: (map(.version) | unique), segments: length}'

# Confirm the supervisor and its offset guard
curl -s "http://<overlord-host>:8090/druid/indexer/v1/supervisor/<datasource>/status" \
  | jq '{state: .payload.state, partitions: .payload.activeTasks}'

Two distinct versions both used for the same hour, or a streaming segment and a batch segment coexisting in the window, is the overlap. A quick correctness cross-check is to compare a COUNT(*) (or a longSum metric) for the overlap window against a known-good total — double the expected value confirms it. The temporal alignment that makes a supervisor and a backfill agree on where "live" ends and "historical" begins is the watermark discipline in the parent guide, synchronizing batch and streaming ingestion; the underlying Kafka wiring is covered in Kafka to Druid real-time pipeline setup.

Target Spec & Validated JSON #

Preventing overlap is a two-sided contract. The streaming side reserves the recent tail for itself with skipOffsetFromLatest, so the supervisor never finalizes segments inside the window a backfill will own. The batch side claims exactly the backfilled window with an explicit intervals plus dropExisting: true, so its handoff atomically retires every existing segment there — streaming or prior batch — and replaces them in one version switch.

The Kafka supervisor spec, with the offset guard that keeps it clear of the backfill window:

{
  "type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "events_web",
      "timestampSpec": { "column": "event_ts", "format": "iso" },
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "MINUTE",
        "rollup": true
      }
    },
    "ioConfig": {
      "type": "kafka",
      "topic": "events_web",
      "consumerProperties": { "bootstrap.servers": "kafka:9092" },
      "taskCount": 2,
      "replicas": 1,
      "taskDuration": "PT1H"
    },
    "tuningConfig": {
      "type": "kafka",
      "maxRowsInMemory": 1000000,
      "intermediatePersistPeriod": "PT10M",
      "skipOffsetFromLatest": "PT1H"
    }
  }
}

The bounded batch backfill that replaces the streaming segments in its window without touching anything outside it:

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "events_web",
      "timestampSpec": { "column": "event_ts", "format": "iso" },
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "MINUTE",
        "rollup": true,
        "intervals": ["2025-11-13T00:00:00Z/2025-11-14T00:00:00Z"]
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "prefixes": ["s3://lake/events/web/2025-11-13/"] },
      "inputFormat": { "type": "json" },
      "appendToExisting": false,
      "dropExisting": true
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsInMemory": 1000000,
      "forceGuaranteedRollup": true,
      "partitionsSpec": {
        "type": "hashed",
        "targetRowsPerSegment": 5000000
      }
    }
  }
}

The three load-bearing fields:

• tuningConfig.skipOffsetFromLatest (streaming) — the supervisor will not create or publish segments within this offset of the latest event, reserving the recent window so a backfill and the stream never fight over the same hour. Set it wider than your backfill lag so the boundary always sits in already-finalized territory.
• granularitySpec.intervals (batch) — bounds the task to exactly the backfilled window. Without it, dropExisting has no defined region to replace and Druid rejects the pairing.
• ioConfig.dropExisting: true (batch) — on handoff, atomically marks every existing segment in intervals unused and switches queries to the new version, so no interval is ever served by two used sets. This is the exactly-once replace primitive; the broader backfill treatment is in dynamic ingestion spec generation. Full field semantics are in the official Apache Druid ingestion spec reference.

Python Automation Script #

The guard below refuses to submit a backfill whose interval extends into the supervisor's reserved tail, then submits and waits for handoff so the replace completes atomically. It reads the supervisor's skipOffsetFromLatest, computes the earliest timestamp the stream still owns, and blocks any backfill that crosses it. It uses only the standard library plus requests, with capped exponential backoff on every cluster call.

import sys
import time
import requests
from datetime import datetime, timedelta, timezone

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


def _iso_to_dt(ts: str) -> datetime:
    return datetime.fromisoformat(ts.replace("Z", "+00:00"))


def stream_reserved_start(datasource: str, skip: timedelta) -> datetime:
    """The earliest instant the live supervisor still owns (now - skip)."""
    # skipOffsetFromLatest reserves the window [now - skip, now] for streaming.
    return datetime.now(timezone.utc) - skip


def assert_no_overlap(interval: str, reserved_start: datetime) -> None:
    """Block a backfill whose end crosses into the streaming-reserved tail."""
    _, end = interval.split("/")
    if _iso_to_dt(end) > reserved_start:
        raise SystemExit(
            f"OVERLAP BLOCKED: backfill end {end} crosses the streaming-reserved "
            f"boundary {reserved_start.isoformat()}; widen skipOffsetFromLatest "
            f"or shrink the backfill window"
        )


def submit_with_backoff(spec: dict, retries: int = 5) -> str:
    delay = 2.0
    for attempt in range(retries):
        try:
            r = requests.post(f"{OVERLORD}/druid/indexer/v1/task",
                              json=spec, timeout=30)
            r.raise_for_status()
            return r.json()["task"]
        except requests.RequestException:
            if attempt == retries - 1:
                raise
            time.sleep(delay)
            delay = min(delay * 2, 30.0)  # capped backoff
    raise RuntimeError("unreachable")


def wait_for_single_version(datasource: str, interval: str,
                            max_wait: float = 600.0) -> None:
    """Poll until the interval is served by exactly one used version."""
    delay, waited = 5.0, 0.0
    iv_prefix = interval.split("/")[0][:10]
    while waited < max_wait:
        r = requests.get(
            f"{COORDINATOR}/druid/coordinator/v1/datasources/{datasource}/segments",
            params={"full": "true"}, timeout=30,
        )
        r.raise_for_status()
        in_window = [s for s in r.json() if s["interval"].startswith(iv_prefix)]
        versions = {s["version"] for s in in_window}
        if in_window and len(versions) == 1:
            return
        time.sleep(delay)
        waited += delay
        delay = min(delay * 2, 30.0)  # capped backoff
    raise RuntimeError(f"{interval} still served by multiple versions after {max_wait}s")


def run_backfill(spec: dict, datasource: str, interval: str, skip_hours: int) -> None:
    reserved = stream_reserved_start(datasource, timedelta(hours=skip_hours))
    assert_no_overlap(interval, reserved)
    task_id = submit_with_backoff(spec)
    wait_for_single_version(datasource, interval)
    print(f"{task_id}: backfill of {interval} replaced cleanly, one used version")

Two safeguards make the replace safe. assert_no_overlap refuses a backfill that would land inside the streaming-reserved tail, catching the mistake before a slot is consumed rather than after queries start double-counting. wait_for_single_version confirms the dropExisting handoff actually collapsed the window to one version — the observable proof that no two used sets remain. If it times out with multiple versions, the old segments never got retired and the interval is still double-counting.

Verification Steps #

After the backfill completes, confirm exactly one used version serves the window. Re-read the timeline for the interval:

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

Expected output shows a single version per hour:

{ "interval": "2025-11-13T00:00:00.000Z/2025-11-13T01:00:00.000Z", "versions": 1 }

Cross-check the count for the overlap window against the known-good total — it should now match, not double:

curl -s -X POST "http://<broker-host>:8082/druid/v2/sql" \
  -H "Content-Type: application/json" \
  -d '{"query": "SELECT COUNT(*) AS c FROM events_web WHERE __time >= TIMESTAMP '\''2025-11-13 00:00:00'\'' AND __time < TIMESTAMP '\''2025-11-14 00:00:00'\''"}'
# [{"c": 4821330}]   # matches source count, not 2x

Finally, confirm the supervisor is healthy and its offset guard is still keeping it off the backfilled window:

curl -s "http://<overlord-host>:8090/druid/indexer/v1/supervisor/events_web/status" \
  | jq '.payload.state'
# "RUNNING"

Enforce these as gates: block any backfill whose interval crosses the streaming-reserved boundary, require a single-version check after every dropExisting handoff, and alert if a datasource ever shows two used versions for one interval. Reference the Druid Kafka ingestion documentation for skipOffsetFromLatest semantics and supervisor handoff timing.

Related #

• Synchronizing batch and streaming ingestion — parent guide: watermark alignment and the full contract for keeping batch and streaming in sync.
• Kafka to Druid real-time pipeline setup — how the supervisor whose tail this guide reserves is configured and run.
• Dynamic ingestion spec generation — the dropExisting + bounded-intervals replace primitive as the basis for deterministic backfills.

Up one level: Synchronizing batch and streaming ingestion frames how batch backfills and live streams share one timeline without conflict.