Tutorial: Kafka Streaming

Goal: run BlazeRules as a streaming operator — consume records from a Kafka topic in batches, evaluate them, and publish the decisions to an output topic.

🚧

Requires the IO module

The release wheel and default source build include blazerules_io with Kafka enabled. Confirm at runtime with blazerules_io.has_kafka, especially if you are using a custom lean build. See Streaming & IO.

Prerequisites

• blazerules and blazerules_io on PYTHONPATH, built with IO + Kafka.
• A running Kafka broker and an input topic with JSON records.
• A rules.yaml.

Step 1 — Load the engine and check Kafka

import json
import blazerules, blazerules_io

assert blazerules_io.has_kafka, "Kafka support is not available in this build"

engine = blazerules.RuleEngine()
engine.load_rules("rules.yaml")

Step 2 — Create a consumer and producer

consumer = blazerules_io.KafkaConsumer(
    brokers="localhost:9092",
    group_id="blazerules-workers",
    topics=["transactions"],
    conf={"enable.auto.commit": "false"},
)

producer = blazerules_io.KafkaProducer(
    brokers="localhost:9092",
    conf={},
)

Step 3 — Consume, evaluate, produce

Poll a batch, evaluate it in one call, then publish per-record decisions.

while True:
    messages = consumer.poll_batch(max_messages=8192, timeout_ms=200)
    if not messages:
        continue

    payload = b"\n".join(messages) + b"\n"
    result = engine.evaluate_ndjson(payload)

    for i, decision in enumerate(result.decisions):
        producer.produce("decisions", value=decision_record(i, decision))

    producer.flush()
    consumer.commit()

📘

Prefer the native loop for throughput

For maximum throughput the IO module also exposes run_stream(engine, config), which runs the consume → evaluate → produce loop inside the native runtime. See Streaming & IO for the StreamRunConfig fields.

poll_batch(...) is a JSON-payload convenience path. For a fully native loop, run_stream(...) and blazerules stream kafka support JSON/NDJSON, Debezium, Arrow IPC, Avro, and Protobuf payloads through StreamRunConfig.payload_format.

Metadata-preserving variant

Use poll_records(...) when you need topic, partition, offset, key, and timestamp metadata in the output.

records = consumer.poll_records(max_messages=8192, timeout_ms=200)
payload = b"\n".join(r.value for r in records) + b"\n"
result = engine.evaluate_ndjson(payload)

for i, decision in enumerate(result.decisions):
    source = records[i]
    out = {
        "source_topic": source.topic,
        "partition": source.partition,
        "offset": source.offset,
        "decision": decision,
        "score": result.scores[i],
        "winning_rule_id": result.winning_rule_ids[i],
    }
    producer.produce("decisions", value=json.dumps(out), key=source.key)

Dead-letter routing and broker security

The native loop (run_stream / blazerules stream kafka) adds two production controls:

• Dead-letter routing. Set --dlq-topic (CLI) or StreamRunConfig.dlq_topic (Python) to route records that fail to decode to a dead-letter topic and keep consuming, instead of aborting the loop. The count of routed records is reported as dlq_routed in the stats/output JSON (alongside batches, messages, matched, emitted, eval_ms/eval_us, and delivery_errors).
• Arbitrary librdkafka settings. Pass repeatable --consumer-conf k=v / --producer-conf k=v (CLI) or the consumer_conf / producer_conf dicts (Python) for SASL/SSL and any other librdkafka setting.

blazerules stream kafka \
  --rules rules.yaml \
  --brokers broker:9093 \
  --input-topic transactions \
  --output-topic decisions \
  --dlq-topic transactions.dlq \
  --consumer-conf security.protocol=SASL_SSL \
  --consumer-conf sasl.mechanism=PLAIN \
  --consumer-conf sasl.username=svc \
  --consumer-conf sasl.password=secret \
  --producer-conf security.protocol=SASL_SSL

Expected output

Each input record yields one decision record on the output topic. A consumer on decisions sees a steady stream keyed/labelled by the engine's verdict (APPROVE, REVIEW, BLOCK, …).

Validation

• The output topic receives exactly result.n_records decisions per processed batch.
• consumer.commit() advances the group offset only after a batch is produced — at-least-once delivery.
• Watch result.messages_skipped for malformed input.

Cleanup

Close the consumer and flush the producer on shutdown so no decisions are lost.

Where to go next