Reading co-partitioned inputs

A named input mapping lets one stateful handler read several Kafka topics. For partition p, the handler receives partition p of every input and shares one state store across them. Each batch contains records from exactly one input, identified by ctx.input; the inputs can have different schemas and event-time paths.

from turbine import KafkaBroker, RecordBatch, Turbine, TurbineContext, TurbineState

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

@app.subscribe(
    input={
        "scored": kafka.topic("wiki.edits.scored", event_time="rev_dt"),
        "tags": kafka.topic("wiki.revert-tags", event_time="dt"),
    },
    partition_key="rev_id",
    partition_hash="murmur2",
)
def verify(batch: RecordBatch, state: TurbineState, ctx: TurbineContext) -> None:
    for row in batch.to_pylist():
        revision = state.scope(str(row["rev_id"]))
        if ctx.input == "scored":
            revision.put("prediction", row)
        else:
            revision.put("tag", row)

app.run()

This example stores the two record shapes separately. Matching them, suppressing duplicates, scheduling expiry and producing verification results are application logic. A class can use process_key(key, rows, state, ctx) to receive an already-scoped state view and on_timer(timer_id, state) to handle expiry. Register an event-time deadline with state.register_event_time_timer(id, fire_time_ms) and cancel it through the same scoped view when a match arrives.

Placement and identity

Declare at least two distinct topics, using nonempty input names. The first input is the anchor: keep its position and every input's name stable across nodes and restarts. A worker moves all its inputs together in cluster mode. Other subscriptions may independently read any of these topics.

Every producer must partition on the declared partition_key with murmur2, using the same serialized key bytes. For librdkafka producers, set partitioner=murmur2_random. Keys must be non-null strings or Int32/Int64 values. All topics must have the same partition count. A confirmed placement mismatch or a partition-count change halts the affected worker. Multi-input state cannot be rescaled or converted from existing single-input state; start with fresh state when changing its inputs or counts.

During an app run, only that multi-input subscription's assignments remain limited to its original partition count. Other subscriptions keep discovering partitions and rebalancing, including a single-input reader of the same topic. An unavailable metadata response is retried; it does not prove that the partition count changed.

Each input keeps its own schema, event time and deserialization on_error/dlq policy. Every input needs event_time or with_kafka_timestamp=True. Both clocks are presented to the handler as _turbine_event_ts in epoch milliseconds. Handler errors follow the subscription's on_processing_error, or the anchor input's on_error when unset. Processing errors on any input use the anchor's dlq destination; secondary inputs need their own DLQ only if their deserialization policy asks for one. DLQ records preserve the raw source payload, key and actual source topic/partition/offset. Destinations must differ from every input topic.

Progress and silence

The subscription's event-time frontier is the minimum of its inputs' watermarks. Each observed input advances from its own data and coasts at real-time speed during silence. Data progress and coasted expiry progress stay separate and survive restarts; downtime does not count as silence. Before all inputs have a watermark, TTL writes receive a conservative timestamp from observed inputs, while expiration still waits for the common frontier. This keeps initial keys expirable once progress resumes.

An input that has never delivered a timestamp blocks the minimum by default. For a sparse label stream, that can prevent expiration forever. Set input_idle_timeout_ms=60_000, for example, to exclude a never-seen input after one minute of observed inactivity. That exclusion survives restart. Its first timestamp makes it participate again. The existing expiry frontier never goes backwards, so those first records can already be late.

For a shared upstream that can stop all inputs together, set input_idle_policy="hold" with input_idle_timeout_ms. The policy applies independently to each partition: after that duration without data on any of its inputs, coasting pauses for that partition. A silent input continues coasting while another input receives data. Resuming any input restarts coasting without adding the paused duration; committed silence survives restart, and application downtime never consumes the remaining timeout. Timers and state TTLs follow the held frontier. The default policy, "coast", keeps advancing even when every input is silent. The detection interval still counts as silence and can expire rows; this option does not reopen completed expiries.

If only a dense input resumes after a shared pause, a sparse input resumes coasting from its held value. The common expiry frontier can therefore remain behind by the paused duration until the sparse input receives a timestamp that lets it catch up. This delays timer-driven output, downstream window closure and state TTL expiry.

The inactivity timeout does not replace an application's verification horizon: a prediction with a 48-hour deadline still waits until that deadline. Choose the timeout with the label feed's potential startup lag in mind. Under "hold", the same timeout also controls outage detection: a longer value waits longer before excluding a never-seen label input, but allows more coasting during a real outage before clocks freeze.

ctx.watermark_ms is the common pre-batch data frontier, or None while blocked. state.close_frontier_ms reports how far timer closure has progressed, including coasting. A hand-written join should account for that expiry frontier when rejecting late labels. Persistent windows use the common frontiers automatically.

Transactions and recovery

Opt into processing_guarantee="exactly_once" with an output and an explicit app_id for Kafka outputs and all input offsets to commit in one transaction. State, timers and per-input progress are journaled together. A local crash reconciles those exact mutations with the committed transaction, so recovery does not need to reproduce the original cross-topic arrival order.

Preserve the worker's local state or a checkpoint matching its sink receipt. If both are lost, an older checkpoint cannot safely reconstruct a potentially order-sensitive handler and startup refuses it. Graceful cluster handoff checkpoints the complete worker before releasing its ownership. The explicit receipt re-enrollment override is a recovery escape hatch with the same loss/duplication risks as for timer output.

In a cluster, a hard node crash can leave the replacement owner with a remote checkpoint older than the last committed batch: every multi-input data batch advances the sink receipt. Automatic reassignment therefore does not guarantee that processing resumes. The partition can remain waiting for handoff, or halt on an incompatible checkpoint once ownership permits startup. Recover compatible local state or a checkpoint, or explicitly re-enroll with the documented recovery risks. Returning the original node with its volume before the rebalance delay expires avoids replacing its state. This restriction applies to the multi-input partition; unrelated subscriptions continue to rebalance.

When re-enrolling empty state against existing sink receipts, on_crash_recovery and halt_if_gap_exceeds still apply to every input. accept resumes at the committed positions with empty application state; replay rebuilds state silently up to those positions before enabling output. These positions take precedence over from_earliest and from_latest because sink history already exists.

Worker lag is the sum across inputs; turbine_input_consumer_lag exposes each input separately and turbine_join_input_skew_ms reports data-watermark skew. The console draws all input edges into one subscription node.

The initial multi-input implementation uses serial polling and decoding. repartition_by, combine_by, infer and max_lag_seconds are currently refused with named inputs. Unsupported declarations fail before processing. Persistent windows, scoped timers, side outputs, checkpoints and cluster ownership use the shared worker lifecycle.

For a reusable persisted interval join with expiry, use interval_join instead of writing the matching state machine by hand.