06 — Kafka-Sourced Computation Graphs

In the previous tutorials, events arrived via WebSocket pushed by an external process. In this tutorial you’ll declare a stream accumulator in package.toml. The server reads events from a Kafka topic automatically — once the graph loads, the accumulator connects to Kafka and pulls messages without any application code changes.

What you’ll learn

The [[metadata.accumulators]] configuration block in package.toml
Setting CLOACINA_VAR_KAFKA_BROKER so the server knows where to connect
Starting Kafka locally with docker compose
Creating topics and producing test messages with kafka-console-producer.sh
Verifying the graph fires after Kafka messages arrive

Prerequisites

Tutorial 04 complete (you know how to package and upload a CG)
Docker and Docker Compose available
curl and python3 available

No special server build is required: the Kafka client ships inside the cloacina-provider-kafka constructor provider your package bundles, not in the server (CLOACI-T-0898).

Time estimate

30–40 minutes

Background

A stream accumulator delivers events to the reactor exactly like a WebSocket accumulator, except a background Kafka reader feeds messages into the channel for you. For how accumulators buffer and deliver events, see Accumulator.

Step 1: Start Kafka

The Cloacina development environment includes Kafka (Apache Kafka 3.9 in KRaft mode — no ZooKeeper). Start it:

# From the Cloacina repository root
docker compose -f .angreal/docker-compose.yaml up -d kafka

Wait for the health check to pass:

docker compose -f .angreal/docker-compose.yaml ps

You should see cloacina-kafka with status healthy. This usually takes 20–30 seconds on first start.

Verify it’s accepting connections:

docker exec cloacina-kafka \
  /opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server localhost:9092

If you see a list of API versions, Kafka is ready.

Step 2: Create the Kafka topic

docker exec cloacina-kafka \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create \
  --topic price.orderbook \
  --partitions 1 \
  --replication-factor 1 \
  --if-not-exists

Expected output:

Created topic price.orderbook.

Step 3: Set CLOACINA_VAR_KAFKA_BROKER

The server resolves broker URLs through the CLOACINA_VAR_ convention. The accumulator’s broker config key names the variable to look up:

export CLOACINA_VAR_KAFKA_BROKER="localhost:9092"

If you’re running the server as a system service, add this to the service environment file. The variable must be set before the graph is loaded — changing it after loading has no effect on already-running accumulator tasks.

Step 4: Write `package.toml` with a stream accumulator

Create a new project directory:

mkdir kafka-price-signal
cd kafka-price-signal

Write package.toml:

[package]
name = "kafka-price-signal"
version = "0.1.0"
interface = "cloacina-workflow-plugin"
interface_version = 1
extension = "cloacina"

[metadata]
graph_name = "kafka_price_signal"
language = "rust"
description = "Price signal graph driven by a Kafka topic"

[[metadata.accumulators]]
name = "orderbook"
accumulator_type = "stream"

[metadata.accumulators.config]
provider = "cloacina-provider-kafka"
constructor = "kafka_source"
broker = "{{ KAFKA_BROKER }}"
topic = "price.orderbook"
group = "kafka-price-signal-group"

[metadata.providers]
cloacina-provider-kafka = "0.1"

A stream accumulator’s source is a constructor provider the package bundles — the Kafka client ships inside cloacina-provider-kafka, not the server (see Consume a Constructor Provider). The [[metadata.accumulators]] array table declares each accumulator. Fields:

Field	Required	Meaning
`name`	Yes	Must match the accumulator name in the graph macro
`accumulator_type`	Yes	`"passthrough"` (WebSocket) or `"stream"` (provider-sourced)
`config.provider`	Yes (stream)	The bundled provider that supplies the source (declared under `[metadata.providers]`)
`config.constructor`	Yes (stream)	The provider member to instantiate (`kafka_source`)
`config.broker`	Yes (stream)	Broker URL or a `{{ VAR }}` template (resolved from `CLOACINA_VAR_{VAR}`)
`config.topic`	Yes (stream)	Kafka topic to consume from
`config.group`	Yes (stream)	Consumer group ID

Multiple accumulators

You can mix passthrough and stream accumulators in the same graph. For example, one accumulator could receive WebSocket pushes while another pulls from a Kafka topic. Add another [[metadata.accumulators]] block for each additional accumulator.

Step 5: Write `Cargo.toml`

[package]
name = "kafka-price-signal"
version = "0.1.0"
edition = "2021"

[dependencies]
cloacina-workflow = { version = "0.10", features = ["packaged", "macros"] }
cloacina-workflow-plugin = "0.10"
cloacina-macros = "0.10"
cloacina-computation-graph = "0.10"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

No [lib] crate-type, no [features] section, no build.rs — the cloacina-compiler service injects the cdylib crate-type and the packaged feature at build time (see 03 — Packaged Workflows).

Step 6: Write `src/lib.rs` — passthrough pattern

The simplest pattern: each Kafka message is deserialized as-is and forwarded to the reactor. The reactor fires on every message (because reaction_mode = "when_any").

use serde::{Deserialize, Serialize};

// One invocation per package — emits the FFI plugin shell.
cloacina_workflow_plugin::package!();

/// Each Kafka message must be a JSON object matching this struct.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OrderBook {
    pub best_bid: f64,
    pub best_ask: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PriceSignal {
    pub mid_price: f64,
    pub spread: f64,
    pub spread_bps: f64,
}

#[cloacina_macros::reactor(
    name = "kafka_price_signal_reactor",
    accumulators = [orderbook],
    criteria = when_any(orderbook),
)]
pub struct KafkaPriceSignalReactor;

#[cloacina_macros::computation_graph(
    trigger = reactor("kafka_price_signal_reactor"),
    graph = {
        compute(orderbook) -> emit,
    }
)]
pub mod kafka_price_signal {
    use super::*;

    pub async fn compute(orderbook: Option<&OrderBook>) -> PriceSignal {
        match orderbook {
            Some(ob) => {
                let mid = (ob.best_bid + ob.best_ask) / 2.0;
                let spread = ob.best_ask - ob.best_bid;
                PriceSignal {
                    mid_price: mid,
                    spread,
                    spread_bps: (spread / mid) * 10_000.0,
                }
            }
            None => PriceSignal {
                mid_price: 0.0,
                spread: 0.0,
                spread_bps: 0.0,
            },
        }
    }

    pub async fn emit(signal: &PriceSignal) -> String {
        format!(
            "mid={:.4} spread={:.4} ({:.2} bps)",
            signal.mid_price, signal.spread, signal.spread_bps
        )
    }
}

Step 7: Package and upload

cd ..
cloacinactl package validate kafka-price-signal
cloacinactl package pack kafka-price-signal
# kafka-price-signal/kafka-price-signal.cloacina

BASE_URL="http://localhost:8080"
TOKEN="clk_your_token_here"

curl -s -w "\nHTTP %{http_code}\n" \
  -X POST "${BASE_URL}/v1/tenants/public/workflows" \
  -H "Authorization: Bearer ${TOKEN}" \
  -F "file=@kafka-price-signal/kafka-price-signal.cloacina;type=application/octet-stream"

Wait for compilation (60–120 seconds on first build):

for i in $(seq 1 30); do
  result=$(curl -s "${BASE_URL}/v1/health/graphs" \
    -H "Authorization: Bearer ${TOKEN}")
  if echo "$result" | python3 -c "import sys,json; d=json.load(sys.stdin); exit(0 if any(r['name']=='kafka_price_signal' for r in d['items']) else 1)" 2>/dev/null; then
    echo "Graph loaded!"
    echo "$result" | python3 -m json.tool
    break
  fi
  echo "Waiting... ($i/30)"
  sleep 5
done

Step 8: Produce messages and verify the graph fires

Use kafka-console-producer.sh inside the container to send a test event. Each line is one Kafka message.

echo '{"best_bid": 100.10, "best_ask": 100.15}' | \
  docker exec -i cloacina-kafka \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic price.orderbook

After a short delay (the Kafka consumer poll interval is at most a few hundred milliseconds), verify the reactor fired:

curl -s "${BASE_URL}/v1/health/graphs/kafka_price_signal" \
  -H "Authorization: Bearer ${TOKEN}" | python3 -m json.tool

Expected:

{
  "name": "kafka_price_signal",
  "health": {
    "state": "running"
  },
  "accumulators": ["orderbook"],
  "paused": false,
  "fires": 1,
  "last_fired_at": "2026-08-01T12:34:56Z"
}

The fires counter increments on every graph fire. For per-fire records (inputs, outputs, duration), list the reactor’s recent fires:

curl -s "${BASE_URL}/v1/health/reactors/kafka_price_signal_reactor/fires" \
  -H "Authorization: Bearer ${TOKEN}" | python3 -m json.tool

Produce several more messages and watch the counter increment:

for i in $(seq 1 10); do
  bid=$(python3 -c "import random; print(round(100 + random.uniform(-0.5, 0.5), 4))")
  ask=$(python3 -c "import random; b=${bid}; print(round(b + random.uniform(0.01, 0.20), 4))")
  echo "{\"best_bid\": ${bid}, \"best_ask\": ${ask}}" | \
    docker exec -i cloacina-kafka \
    /opt/kafka/bin/kafka-console-producer.sh \
    --bootstrap-server localhost:9092 \
    --topic price.orderbook
  sleep 0.5
done

Variations

This tutorial covered the passthrough path, where each Kafka message fires the graph. Other accumulators can batch, deduplicate, or combine multiple sources before firing — pick one based on your event semantics. See Choosing Accumulator Types for the decision guide, and the Accumulator reference for the underlying primitive (including consumer-group offset and restart behavior).

Troubleshooting

Accumulator shows a degraded state and graph never fires: The Kafka connection failed. Check the server logs for provider stream accumulator FAILED or kafka_source[...]: poll error messages. Verify CLOACINA_VAR_KAFKA_BROKER is set correctly and that the broker is reachable from the server process. If running the server inside a container, localhost:9092 may not resolve correctly — use the Docker network hostname instead (e.g., cloacina-kafka:9092).

Messages produce but the graph’s fires counter stays at 0: The message payload is not valid JSON matching your boundary type. Verify with kafka-console-consumer.sh:

docker exec cloacina-kafka \
  /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic price.orderbook \
  --from-beginning \
  --max-messages 5

stream accumulator declares no provider error in server logs: the accumulator config is missing the provider/constructor routing keys — add them (and the [metadata.providers] declaration) as shown in Step 4. Kafka support ships in the bundled provider, not the server, so there is no server feature flag to enable.

Topic does not exist: the provider’s consumer logs a poll error and keeps retrying (the stream stays alive) — the topic is picked up as soon as it exists. Auto-topic-creation on the first produced message also works with the default broker config.

Next steps

Next: 07 — Cross-Package Reactor Binding