Kafka Workflows

@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 has the full breakdown.

pip install "pyrula[workflows,kafka]"

Bind a workflow to topics

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

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.

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:

@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.