Worked examples

Two complete Kafka connector configurations with the workflows that consume them - order events given a lane per order, and call events threaded into one session across regional topics.

Two configurations taken end to end, each with the workflow it feeds. Both assume you have read Configuration; neither needs anything from the other.

Order events, one order at a time

An e-commerce team publishes order lifecycle events to orders.events, keyed by order id, with the event type inside the payload. They want no two events for the same order applied at once, different orders handled in parallel, and no run at all for a page view.

duraton:
  app: orders
  topics: [orders.events]
  decoder: json
  envelope: wrapped
  filter: 'event.data.value.type != "order.viewed"'
  mapping:
    eventName: value.type
    tags:
      region: value.region

kafka:
  bootstrap.servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
  group.id: duraton-orders
  auto.offset.reset: latest
  isolation.level: read_committed

eventName: value.type means one topic feeds many workflows: a record whose payload says "type": "order.paid" triggers whatever subscribes to order.paid.

Giving each order a lane of one is a line of workflow config. With the wrapped envelope the Kafka record key is at key, so concurrency keyed on it admits one run per order at a time while different orders run fully parallel:

type OrderEnvelope = {
  key: string;
  offset: number;
  value: { type: string; orderId: string; total: number };
};

export const orderPaid = defineWorkflow<OrderEnvelope>({
  name: "order.paid",
  concurrency: { limit: 1, key: "key" },
  handler: async (ctx) => {
    const { key: orderId, value: order, offset } = ctx.event.data;
    ctx.log.info("applying order event", { orderId, offset });

    const invoice = await ctx.step.run("invoice", () => raiseInvoice(order));
    await ctx.step.run("notify", () => emailCustomer(orderId, invoice));
    return invoice;
  },
});

from duraton import ConcurrencyConfig, define_workflow
from duraton.context import StepContext


async def handle_order_paid(ctx: StepContext) -> object:
    envelope = ctx.event.data
    order_id = envelope["key"]
    order = envelope["value"]
    ctx.log.info("applying order event", {"orderId": order_id, "offset": envelope["offset"]})

    invoice = await ctx.step.run("invoice", lambda: raise_invoice(order))
    await ctx.step.run("notify", lambda: email_customer(order_id, invoice))
    return invoice


order_paid = define_workflow(
    "order.paid",
    handle_order_paid,
    concurrency=ConcurrencyConfig(limit=1, key="key"),
)

type OrderEnvelope struct {
	Key    string `json:"key"`
	Offset int64  `json:"offset"`
	Value  Order  `json:"value"`
}

var orderPaid = duraton.DefineWorkflow(duraton.WorkflowDefinition{
	Name:        "order.paid",
	Concurrency: &duraton.ConcurrencyConfig{Limit: 1, Key: "key"},
	Handler: func(c *duraton.Context) (any, error) {
		envelope, err := duraton.EventData[OrderEnvelope](c)
		if err != nil {
			return nil, err
		}
		c.Log.Info("applying order event", map[string]any{
			"orderId": envelope.Key,
			"offset":  envelope.Offset,
		})

		invoice, err := duraton.Run(c, "invoice", func() (Invoice, error) {
			return raiseInvoice(envelope.Value)
		})
		if err != nil {
			return nil, err
		}
		return duraton.Run(c, "notify", func() (any, error) {
			return nil, emailCustomer(envelope.Key, invoice)
		})
	},
})

The filter drops order.viewed before any event is sent, so the noisiest record on the topic never becomes a run and never costs anything.

A concurrency key buys exclusion, not sequence. One order's runs never overlap, but two runs already queued behind the same key are admitted in whatever order the queue offers them, not in the order the partition delivered them. A refused run is simply retried a moment later, and nothing tiebreaks on creation time.

That is enough for work that must not collide (reserving stock while capturing payment against the same order) and not enough for work that must be applied in sequence (a state machine that rejects order.shipped before order.paid). When sequence is the requirement, make the workflow itself tolerate arriving early: read the current state, and park or reject rather than assume the previous event has landed. The record's offset is in the wrapped envelope precisely so a workflow can tell how far along the partition an event sat.

Call events across regional topics

A voice platform writes call lifecycle events to one topic per region, over a mutually authenticated cluster. Records are not keyed, so ordering has to come from the payload. Runs for one call should thread into a single session.

duraton:
  app: voice
  topicPattern: '^voice\.calls\.(eu|us|apac)$'
  decoder: json
  envelope: wrapped
  mapping:
    eventName: value.event
    session: value.callId
    tags:
      region: value.region
      trunk: value.trunk

kafka:
  bootstrap.servers: kafka-1.voice.internal:9093,kafka-2.voice.internal:9093
  group.id: duraton-voice
  auto.offset.reset: earliest
  security.protocol: SASL_SSL
  sasl.mechanism: SCRAM-SHA-512
  sasl.username: duraton
  ssl.ca.location: /etc/duraton/ca.pem
  max.poll.records: 200

Two things differ from the first example beyond the domain. The subscription is a pattern, and it is anchored: ^voice\.calls\.(eu|us|apac)$ matches those three names and nothing else, so a new voice.calls.dlq topic does not silently join the subscription. And because the producer does not set a record key, per-call ordering is keyed on a payload path instead:

type CallEnvelope = {
  topic: string;
  value: { event: string; callId: string; region: string; durationMs: number };
};

export const callCompleted = defineWorkflow<CallEnvelope>({
  name: "call.completed",
  concurrency: { limit: 1, key: "value.callId" },
  handler: async (ctx) => {
    const { callId, durationMs } = ctx.event.data.value;
    ctx.log.info("rating call", { callId, topic: ctx.event.data.topic });

    const rated = await ctx.step.run("rate", () => rateCall(callId, durationMs));
    await ctx.step.run("post-cdr", () => postCDR(rated));
    return rated;
  },
});

from duraton import ConcurrencyConfig, define_workflow
from duraton.context import StepContext


async def handle_call_completed(ctx: StepContext) -> object:
    envelope = ctx.event.data
    call = envelope["value"]
    ctx.log.info("rating call", {"callId": call["callId"], "topic": envelope["topic"]})

    rated = await ctx.step.run("rate", lambda: rate_call(call["callId"], call["durationMs"]))
    await ctx.step.run("post-cdr", lambda: post_cdr(rated))
    return rated


call_completed = define_workflow(
    "call.completed",
    handle_call_completed,
    concurrency=ConcurrencyConfig(limit=1, key="value.callId"),
)

type CallEnvelope struct {
	Topic string `json:"topic"`
	Value Call   `json:"value"`
}

var callCompleted = duraton.DefineWorkflow(duraton.WorkflowDefinition{
	Name:        "call.completed",
	Concurrency: &duraton.ConcurrencyConfig{Limit: 1, Key: "value.callId"},
	Handler: func(c *duraton.Context) (any, error) {
		envelope, err := duraton.EventData[CallEnvelope](c)
		if err != nil {
			return nil, err
		}
		c.Log.Info("rating call", map[string]any{
			"callId": envelope.Value.CallID,
			"topic":  envelope.Topic,
		})

		rated, err := duraton.Run(c, "rate", func() (Rating, error) {
			return rateCall(envelope.Value.CallID, envelope.Value.DurationMS)
		})
		if err != nil {
			return nil, err
		}
		return duraton.Run(c, "post-cdr", func() (any, error) {
			return nil, postCDR(rated)
		})
	},
})

session: value.callId threads every run started by that call - setup, answer, completion - into one session, so the console shows a call as one conversation rather than as three unrelated runs. max.poll.records: 200 is lowered from the default because these records are large; the startup poll-window check will tell you if the number you pick cannot hold.

Filtering - the CEL dialect both filters are written in.
Security - the SASL and TLS properties in the second example.
Behaviour worth knowing - why that pattern is anchored.

Order events, one order at a time

Call events across regional topics

Next

On this page