Skip to main content

kafka-pipeline template

Kafka source consuming orders.created, enrichment via Haiku, re-emit to orders.enriched, DLQ on orders.dlq. Declares dailyTokenUSD: 5 to exercise Phase-6 cost enforcement.

Scaffold​

declaragent init --template kafka-pipeline --provider anthropic

Ships with a Redpanda docker-compose.yaml for the local dev loop.

Canonical starter: templates/kafka-pipeline/.

Run locally​

cd kafka-pipeline
docker compose up -d                       # Redpanda single-broker
declaragent up -d                          # bind kafka source + start the pipeline
# produce a test event
echo '{"id":"ord_1","total":75,"items":[{"sku":"BW","name":"Blue Widget","quantity":3}],"shipTo":{"country":"US","city":"Austin","region":"TX"}}' \
  | docker compose exec -T redpanda rpk topic produce orders.created

Expected flow:

  1. Source adapter consumes orders.created.
  2. Engine runs the enrich skill.
  3. Engine emits the enriched JSON to orders.enriched.
  4. On error, the message goes to orders.dlq with the failure metadata.

Key points​

  • DLQ. Every source declares its DLQ topic. See declaragent dlq list / show / redrive.
  • Cost budget. dailyTokenUSD: 5 trips EQUOTA when exceeded. The source pauses consumption until the next UTC rollover.
  • Idempotency. The payload's $.id (via routing.idempotencyKey.path) is the idempotency key — re-processing the same order is safe on restart.