Interval joins

Join two co-partitioned Kafka topics before your handler. Buffered rows and their expiry index are persisted and checkpointed; the worker commits both input positions with the joined outputs under processing_guarantee="exactly_once". A restart resumes the buffered join without replaying its horizon.

import pyarrow as pa
from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState, interval_join

Prediction = pa.schema([
    pa.field("rev_id", pa.string(), nullable=False),
    pa.field("wiki", pa.string(), nullable=False),
    pa.field("risk", pa.float64(), nullable=False),
    pa.field("predicted_at", pa.string(), nullable=False),
])
Label = pa.schema([
    pa.field("rev_id", pa.string(), nullable=False),
    pa.field("wiki", pa.string(), nullable=False),
    pa.field("labelled_at", pa.string(), nullable=False),
])

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka, app_id="verification")

@app.subscribe(
    interval_join(
        kafka.topic("predictions", schema=Prediction, event_time="predicted_at"),
        kafka.topic("labels", schema=Label, event_time="labelled_at"),
        on=["rev_id", "wiki"], between=(0, 48 * 3_600_000),
        how="left", emit="first", dedupe_left=True, indicator="matched",
        expiry_bucket_ms=60_000,
    ),
    partition_key="rev_id", partition_hash="murmur2",
    output=kafka.topic("verified", message_key="rev_id"),
    processing_guarantee="exactly_once", input_idle_timeout_ms=60_000,
)
def verify(batch: RecordBatch, state: TurbineState) -> RecordBatch:
    return batch

app.run()

Both topics need explicit schemas, their own event-time axes, the same partition count and producers that place the declared string partition_key using murmur2. on includes that column and can add equality columns such as wiki. Join-key types must agree. The partition key must be non-null and fit in 255 UTF-8 bytes; null values in other equality keys never match. The label's timestamp is when the label happened, rather than a copy of the prediction timestamp.

between=(lower, upper) is inclusive: a right row matches when left.time + lower <= right.time <= left.time + upper. Negative bounds support events that precede the left event. Expiry deadlines must fit non-negative signed 64-bit milliseconds; an unrepresentable deadline rejects the batch through the processing-error policy.

Option	Behaviour
how="inner"	Emit matching rows; discard unmatched rows at expiry.
how="left"	Also emit each unmatched left row once at expiry, with null right columns.
emit="pairs"	Emit every matching pair. Duplicates multiply results.
emit="first"	Emit each left row at its first match, or at expiry for a left join. Further matches are counted and ignored.
dedupe_left=False	Opt in with True to drop repeated left rows with the same on values and event time while the original is buffered.
allowed_lateness_ms=0	Tolerance behind the common frontier; expiry waits that extra duration. A left join also probes retained predictions for late right rows.
max_rows_per_key=10_000	Cap buffered rows per equality key and side. Exceeding the cap halts that partition and increments an overflow counter.

By default, first treats repeated left input rows as separate predictions. Set dedupe_left=True for an at-least-once prediction feed: all on values plus the canonical event time identify a prediction, and the first accepted payload wins even if a replay changes other columns. Identical null equality values also identify replays, although they never match a right row. Different event times remain separate predictions. Right rows retain their normal pairs or first behaviour. Replays consume no additional buffer capacity, state keys or timers, and the retained identity survives restart and checkpoint recovery.

Late classification runs before deduplication. After the original row expires, replays follow the normal late policy. With a non-negative upper bound and unchanged lateness, they are already late; a negative upper bound or an increase in lateness can admit them again after retirement. This option does not provide a permanent history of seen predictions or deduplication when a replay changes its event time. An orphan right row produces no output and expires silently.

on_processing_error applies to the handler on both matches and unmatched expiries. fail halts the partition cleanly. For an expiry batch, skip drops its generated rows and dlq sends those joined rows as JSON, including explicit null right columns, to the DLQ declared on the left input. These records have turbine_dlq_phase=processing_expiry, the anchor topic/partition and source offset -1, because no single source record represents a generated expiry. They contain joined data for inspection, rather than a raw input payload. Failed handler writes and queued outputs are rolled back. Under exactly-once, retiring the buffered rows and publishing their DLQ records are atomic. Absorbed failures update processing-error health and the DLQ counters.

For handler errors during input processing, dlq also uses the left input's destination, whichever input triggered the handler. These records contain the raw source payload with turbine_dlq_phase=processing and the actual source topic, partition and offset. Without an explicit on_processing_error, the handler policy inherits the left input's on_error for both input processing and expiry. Deserialization errors use each input's own on_error and dlq settings.

The left buffer lives through left.time + upper + allowed_lateness_ms; the right buffer through right.time - lower + allowed_lateness_ms. The inclusive final millisecond remains eligible. Expiry rounds the following millisecond up to expiry_bucket_ms. By default, the bucket width is approximately one sixty-fourth of the larger retention horizon, including lateness. Setting a smaller width reduces extra emission delay and increases timer memory. A 48-hour horizon defaults to roughly 45-minute buckets; the example explicitly chooses at most one minute of rounding delay. The common watermark must also advance past the deadline.

An input that has never delivered a row holds the watermark by default. input_idle_timeout_ms explicitly excludes it after inactivity, so unmatched predictions can expire even on a partition with no labels. Its first records can then be late. Watermark clocks resume after restart without counting downtime. In a left join, a right row behind the tolerance still matches any retained left rows. It is buffered only if a future on-time left row could match it (right.time - lower >= frontier - allowed_lateness_ms). A late right row with neither possibility is dropped and counted, or sent in its original input shape through on_late="route", late_data=kafka.topic("late").

For inputs sharing an upstream, set input_idle_policy="hold" alongside input_idle_timeout_ms. Independently in each partition, expiry clocks stop advancing after every input has been silent for that duration. When any input resumes, its silent peers resume coasting without counting the paused duration. If only the dense input resumes, a sparse input's held clock can delay unmatched results and downstream window closure until its timestamps catch up. Silence already observed before a restart remains observed; application downtime adds nothing. A single silent input still coasts while another receives data. The default input_idle_policy="coast" keeps advancing during a shared upstream outage. Neither policy retracts completed expiries, and the detection interval can still close predictions near their deadline. A longer timeout accommodates more label startup lag but also spends more of the verification horizon before freezing during an outage.

The handler accepts state and receives joined batches, including expiry batches from timers. It can use persistent windows, return a main output or call ctx.emit for side outputs. Shared equality columns appear once. Other name collisions use suffixes=("_left", "_right"); indicator="matched" adds a bool that distinguishes an unmatched result from a matching right row whose payload is null. The canonical event time is the later source timestamp for a match and the firing frontier for an unmatched expiry. Windows inherit that clock; configure their lateness tolerance to retain delayed matches when needed. This measures verification-time aggregates. Aggregate by an original prediction timestamp in a separate subscription with its own clock if that is the desired cohort.

The join runs in cluster mode, and parallelism redistributes expiry results to each key's handler shard. Expiry handler calls currently run serially, so increasing parallelism does not speed that path. The current multi-input boundaries still apply: no rescaling, repartition hops, combining, inference or max_lag_seconds. Unsupported declarations are refused before processing.

You can change dedupe_left, allowed_lateness_ms, max_rows_per_key, expiry_bucket_ms, suffixes and indicator across restarts while preserving buffered rows. Enabling deduplication suppresses future replays against the existing buffer; it does not merge or remove copies already buffered. Disabling it accepts future copies as separate rows. Neither change requires a new app identity. Raise the capacity limit to resume a partition halted by a hot key; a lowered limit applies to new inserts without truncating existing matching rows. Existing rows retain their scheduled expiry buckets and deadlines, while new rows use the new bucket width and lateness. Raising lateness does not extend existing deadlines or restore retired buffered rows; lowering it does not shorten existing deadlines. The new tolerance applies immediately to incoming rows. You can also change the idle policy and timeout across restarts; the committed close frontier never goes backwards.

Input schemas can also evolve across restarts without losing buffered rows. You can add a field declared optional (pa.field("page_title", pa.string(), nullable=True), or page_title: str | None = None in a Pydantic model), remove a field that is neither an equality key nor the event-time field, or make an existing field optional. Rows buffered before the change read an added field as null until they match or expire, so during one retention horizon a null there can also mean "buffered before the field existed". A removed field leaves the output, including for rows buffered before its removal. A field removed and later added back must keep its type and be optional, since rows buffered in between have no value for it. Compatibility is judged on top-level fields: a change inside a nested type is a type change. Turbine refuses any other schema change at startup and names the field: a changed type, a new required field, a field made required, any change to an equality key or an event-time field, and an added or removed field whose name the other input also uses, because suffixing would rename an existing output column.

Matching/routing semantics (on, between, how, emit, input topics/clocks, partition key and key-group count) remain immutable and are checked at startup. Changing them, or making an incompatible schema change, requires fresh state and checkpoints under a new app identity; the hand-written multi-input state is not migrated.

The topology API and console show the interval, equality keys and emission mode with both input edges. The console’s Timers view shows retained rows per side, pending expiry buckets, the close frontier, late-row counts and input skew for each worker. A matched row can remain buffered until its retention deadline; one expiry bucket can contain many rows. turbine_join_buffered_rows{worker, join, side} reports committed occupancy and is restored after restart. turbine_active_timers{worker, kind="join"} counts its current expiry buckets. Join expiries and user timers are excluded from the Windows badges. Prometheus exposes turbine_join_late_rows_total, turbine_join_ignored_matches_total, turbine_join_duplicate_rows_total{join, side="left"}, turbine_join_overflow_total, match/unmatched counters and the input-skew gauge. These counters count processing attempts, so failures can increment them before a transaction commits.