Side Outputs¶

A handler normally has one output: whatever process() returns goes to the subscription's output= topic. Side outputs add named destinations alongside it, so one handler can split a stream without reading it twice.

@app.subscribe(
    kafka.topic("events", schema=Event),
    output=kafka.topic("events-scored"),
    side_outputs={
        "critical": kafka.topic("alerts-critical", message_key="tenant_id"),
        "audit": kafka.topic("audit-log"),
    },
)
class score:
    def process(self, batch, ctx):
        ctx.emit("audit", batch)
        ctx.emit("critical", batch.filter(pc.greater(batch.column("score"), 90)))
        return scored(batch)

Three things happen there: every row is copied to audit-log, the rows scoring above 90 also go to alerts-critical keyed by tenant, and the handler's return value still goes to events-scored as usual.

That is the imperative form. When the split is a classification — each row belongs to exactly one destination, chosen by a rule — declare it instead with branch=, which is exclusive and shows its rules on the topology page.

Asking for the context¶

A handler receives ctx by naming a parameter ctx. Four shapes are accepted:

def process(self, batch): ...
def process(self, batch, state): ...
def process(self, batch, ctx): ...
def process(self, batch, state, ctx): ...

ctx must come last. Any other second parameter is the state store, whatever you called it — so handlers written before side outputs existed keep working unchanged. Asking for ctx alone does not make the subscription stateful: only state does that.

Emitting¶

ctx.emit(name, batch) takes an Arrow RecordBatch or Table — a slice, not a record. This is deliberate and it is the reason side outputs are cheap: a filter produces a view over the same buffers, so an N-way split costs one vectorised mask per destination rather than N copies of every row. There is no emit_row; a per-record branch is a mask you haven't written yet.

Emitting an empty batch is normal — a branch that matched nothing this batch — and produces nothing.

ctx.side_outputs lists the declared names, if you want to branch on what is configured rather than hardcode.

What is declared, and what that buys¶

Destinations are declared on @app.subscribe, not named inside emit. That is what lets Turbine validate them at startup, show them on the topology page, and give you a clear error instead of a mystery:

An undeclared name raises immediately, naming the branches that are declared. It is never a silent drop, and never a merge into the main output.
A declared branch nobody writes to is fine — a predicate that matched nothing is not an error.
A branch cannot point at the subscription's own input (that loops forever) or at its output= (rejected rows and results would interleave in the same stream). Both are startup errors.
Names are local to the subscription. Two subscriptions may use the same name for different things; one subscription may point two branches at the same topic, and they stay two branches.

Splitting by rule: `branch=`¶

ctx.emit is imperative and non-exclusive: you decide, row group by row group, and the same rows may go to several destinations. When what you actually want is this row goes to exactly one place, decided by a rule, declare the split instead:

from turbine import Branch
from turbine import functions as f

@app.subscribe(
    kafka.topic("events", schema=Event),
    output=kafka.topic("events-normal"),
    branch=[
        Branch("critical", f.col("score") >= 90, to=kafka.topic("alerts-critical")),
        Branch("warning",  f.col("score") >= 70, to=kafka.topic("alerts-warning")),
    ],
)
class score:
    def process(self, batch):
        return scored(batch)

The handler returns one batch, as always. Turbine then routes each of its rows to the first arm whose predicate holds — a row scoring 95 goes to critical and not to warning, even though it satisfies both. That is why branch= is a list and not a mapping: the order decides the outcome, so it is written as an order.

Rows matching no arm take the default arm, which is output=. There is no third place for them to go — so note that adding branch= to an existing subscription narrows what its output topic receives: it now carries the remainder, not everything. That is what makes the split a partition rather than a fan-out, and it is why branch= and ctx.emit are two surfaces.

Saying what happens to the rest¶

If you have no output=, the rows matching no arm would simply disappear — so Turbine refuses to start until you say that is what you meant:

@app.subscribe(
    kafka.topic("events", schema=Event),
    branch=[Branch("critical", f.col("score") >= 90, to=kafka.topic("alerts"))],
    on_unmatched="drop",     # required: there is no output= to fall through to
)

Dropped rows are counted on turbine_branch_rows_dropped_total{worker} — a series that only exists when you asked for the loss, so it waking up always means data is leaving the pipeline. Setting on_unmatched="drop" and an output= is refused: nothing would ever reach the output topic.

Why the predicate is an expression¶

The predicate is a turbine.functions expression rather than a Python function, and the difference is not stylistic: an expression can be read. Each arm's rule is shown on the topology page and served by the management API, so the shape of your split is visible from outside the process. A lambda would route the same rows and tell nobody anything.

A predicate that evaluates to null for a row does not match it. That matters when a predicate can fail — under the default on_compute_error="null" a failing comparison yields null, and those rows fall through to the next arm rather than being captured by a rule that did not actually hold.

Choosing between the two¶

branch= and ctx.emit answer different questions and compose freely on one subscription:

	`branch=`	`ctx.emit`
A row can land in	exactly one destination	any number
The rule is	declared, and visible from outside	code inside your handler
Written in	expressions	Python
Good for	classifying (severity, tier, region)	copying, auditing, ad-hoc slices

Reach for branch= when the split is a classification and you want it legible; reach for ctx.emit when a destination is a copy, or when the rule is genuinely procedural.

Delivery¶

Side outputs are ordinary outputs. Under exactly_once every destination is written inside the same transaction as the batch's input offsets, so a crash cannot leave one branch written and another not, and no branch is ever visible twice. Under at-least-once they are produced with the batch like any other output.

This holds because all destinations live on one Kafka cluster and a Kafka transaction spans every produce in the batch. Fanning out to a different system (a lakehouse table, say) alongside Kafka would have two independent commit points and therefore no joint guarantee — which is why it is not offered under exactly_once rather than offered with a footnote.

Per-destination volume is on turbine_side_output_records_total{worker, destination}, for branches and imperative emits alike.

When to use a side output, and when not to¶

Turbine gives you two ways to get several streams out of one topic, and they are not the same:

	Side outputs (`branch=` / `ctx.emit`)	Several `@app.subscribe` on one topic
Consumers	one	one per subscription
Decode cost	paid once	paid per subscription
Offsets / guarantee	shared	independent
Failure domain	shared — one handler	isolated

Use a side output when the branches are facets of one decision made by one piece of logic (score → alert / audit / result). Use separate subscriptions when the consumers are genuinely independent and you want one to be able to fail, lag, or be redeployed without touching the other — see multiple subscriptions.