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 = "2024"

[dependencies]
cloacina-workflow = { version = "0.12", features = ["packaged", "macros"] }
cloacina-workflow-plugin = "0.12"
cloacina-macros = "0.12"
cloacina-computation-graph = "0.12"
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