# Kafka Workflows

> Bind a durable workflow to Kafka topics: consume, run checkpointed steps, produce, with exactly-once processing.

Source: https://docs.pyrula.io/workflows/guides/kafka/

`@kafka_workflow` runs a durable `@workflow` per Kafka record: consume, decode, checkpointed
steps, encode, produce. Undecodable records go to a dead-letter queue (DLQ) before the workflow
runs.

Processing is exactly-once by default: a redelivered record whose run completed is read back from
the run store, not re-run. Output is at-least-once by default: a crash between produce and offset
commit re-produces on redelivery. Dedupe downstream on key, or set `delivery="exactly_once"` for
atomic output + offset. The [guarantee matrix](https://docs.pyrula.io/workflows/guides/kafka-guarantees.md) has the full
breakdown.

```
pip install "pyrula[workflows,kafka]"
```

## Bind a workflow to topics

```python
from pydantic import BaseModel
from pyrula.workflows.kafka import kafka_workflow, KafkaWorker

class Event(BaseModel):
    id: int
    ip: str

class Enriched(BaseModel):
    id: int
    geo: str

@kafka_workflow(input_topic="events", group="enricher",
                output_topic="events.enriched",
                input_schema=Event, output_schema=Enriched)
async def enrich(ctx, event: Event) -> Enriched:
    geo = await ctx.step("geo", lambda: lookup(event.ip))
    return Enriched(id=event.id, geo=geo)
```

`*_schema` takes a pydantic model (JSON), a wire `Format`
(`Json`/`Avro`/`JsonSchema`/`Protobuf`/`Raw`/`String`), or any `MessageCodec`. Decode failure goes
to the DLQ.

Registry-backed formats read `schema_registry=` on the binding, else
`KafkaConnection.schema_registry_url` (`PYRULA_SCHEMA_REGISTRY_URL`). Requests are bounded at 5 s;
raise with `schema_registry_timeout_ms=` or `PYRULA_SCHEMA_REGISTRY_TIMEOUT_MS` (1 to 300000,
else `ValueError`).

Other binding options:

| Option | Default | |
|---|---|---|
| `on_error` | `"dlq"` | After retries: `dlq`, `fail` or `skip` |
| `dlq_topic` | `<input_topic>.dlq` | DLQ topic |
| `max_retries` | `2` | Exponential backoff before `on_error` |
| `max_concurrency` | `8` | Records in flight |
| `idempotency_window` | `10_000` | Recent runs kept per partition for redelivery dedup |
| `input_schema=list[X]` | | Batch handler: 1..`max_batch_size` (256) records per call; `batch_linger_ms` waits to fill |
| `delivery` | `"at_least_once"` | `"exactly_once"` for transactional output + offset |

## Run it

```python
import asyncio
from pyrula.workflows.kafka import KafkaConnection, KafkaWorker
from pyrula.workflows.stores.valkey import ValkeyStore

worker = KafkaWorker(
    [enrich],
    connection=KafkaConnection.from_env(),   # PYRULA_KAFKA_*
    store=ValkeyStore(...),                   # run state + redelivery dedup
)
asyncio.run(worker.run())
```

The store decides what a restart costs. `ValkeyStore`: a redelivered record reuses its stored
result. `MemoryStore()`: dedup only within this process. A durable binding with no store raises at
construction.

Run id is `group:topic:partition:offset:fp` (`first-last` offsets for a batch), `fp` hashing the
record's timestamp, key, value and headers. A different record at a reused offset is a new run; a
byte-identical redelivery reuses its result. The id rides on every output in the `pyrula.run_id`
header.

```python
from pyrula.workflows.stores.memory import MemoryStore

worker = KafkaWorker([enrich], store=MemoryStore())   # tests, single-process demos
```

## Skip durability

For an idempotent handler, `durable=False` runs it straight through with no run-event store,
dedup or replay:

```python
@kafka_workflow(
    input_topic="events", group="enricher",
    input_schema=Event, output_topic="events.enriched", output_schema=Enriched,
    durable=False,
)
async def enrich(ctx, event: Event) -> Enriched:
    return Enriched(id=event.id, geo=lookup(event.ip))
```

At-least-once: a redelivered record re-runs and re-produces. `ctx.step` runs inline, nothing
persisted. `KafkaWorker` passes `store=` only to durable bindings; `KafkaRunner(store=...)` with
`durable=False` raises, and so does `durable=False` with `delivery="exactly_once"`.

## Straight-through only

A Kafka workflow must finish in one pass. A pause (`ctx.sleep`, `ctx.interrupt`,
`ctx.wait_for_signal`) is an error and follows `on_error`. A retried record resumes from its
checkpointed steps.

## A store outage holds the record

When the run store fails under a record (Valkey restart, dropped connection), the run is requeued
as `store_unavailable` and resumes from its journal. The runner holds the record: it reclaims the
run with backoff from 0.1 s doubling to 5 s, spends no `max_retries` and never reaches `on_error`.
The offset stays uncommitted while earlier records still commit. A batch binding holds the whole
batch.

A stop or partition revoke ends the wait at its next check and leaves the offset for the next
owner. A stop during the requeue retry can take up to the store's `lock_ttl`.

The hold has no deadline, since a dead store takes no record. For a store that's up but keeps
failing the run, the store's `max_store_requeues` (20 default) caps it; then the run ends with
`run:error` code `store_unavailable` and follows `max_retries` and `on_error`.

A hold logs `kafka_record_held` (reason, elapsed seconds, position) once per reason, then every
30 s. Stats count them as `held`; alert on it staying above zero.

## Relation to agents

`@kafka_agent` (in `pyrula.agents.kafka`) is `@kafka_workflow` plus `@agent`: the same delivery,
DLQ, dedup and commit mechanics, with a turn-scoped store and the LLM loop on top.