Recipe: Run the Native Kafka Loop

For maximum throughput, keep the Kafka consume → evaluate → produce loop inside the native IO runtime.

run_stream(...) and blazerules stream kafka can evaluate JSON/NDJSON, Debezium, Arrow IPC, Avro, or Protobuf message values. Binary formats decode to Arrow batches before evaluation; they do not detour through JSON.

The loop itself is a concurrent multi-worker pipeline as of 0.5.3 — one thread polls and dispatches by partition, worker_count workers evaluate in parallel on their own engine shards, and one delivery thread produces, flushes, and commits. See Streaming & IO for the full architecture; the config below tunes it.

import blazerules
import blazerules_io

assert blazerules_io.has_kafka

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

cfg = blazerules_io.StreamRunConfig()
cfg.brokers = "localhost:9092"
cfg.group_id = "blazerules-workers"
cfg.input_topics = ["transactions"]
cfg.output_topic = "decisions"
cfg.batch_size = 8192
cfg.worker_count = 4                # concurrent evaluation workers, each with its own engine shard
cfg.queue_depth = 256                # depth of every bounded queue (per-worker input + delivery)
cfg.poll_timeout_ms = 200
cfg.flush_timeout_ms = 5000
cfg.flush_interval_ms = 250          # delivery thread flush/commit cadence
cfg.commit_offsets = True
cfg.partition_affine = True          # forced True whenever commit_offsets is True
cfg.output_mode = "rows"             # rows, grouped, or none
cfg.payload_format = "json"       # json, ndjson, debezium, arrow-ipc, avro, protobuf
cfg.arrow_validation = blazerules_io.ArrowIpcValidationLevel.STRUCTURAL  # full, structural, trusted
cfg.avro_schema_json = ""          # required for avro
cfg.protobuf_descriptor_set = b""   # required for protobuf
cfg.protobuf_message_type = ""      # required for protobuf

stats = blazerules_io.run_stream(engine, cfg)
print(stats.batches, stats.messages, stats.matched, stats.emitted, stats.eval_us)
print(stats.delivery_errors, stats.dlq_routed)

CLI equivalent

blazerules stream kafka \
  --rules rules.yaml \
  --brokers localhost:9092 \
  --input-topic transactions \
  --output-topic decisions \
  --format json \
  --batch-size 8192 \
  --workers 4 \
  --queue-depth 256

For Avro or Protobuf topics, add --schema or --descriptor / --message.

Set limits for local tests:

cfg.max_messages = 1_000_000
cfg.max_batches = 100

CLI equivalent with limits

blazerules stream kafka \
  --rules rules.yaml \
  --brokers localhost:9092 \
  --input-topic transactions \
  --output-topic decisions \
  --format json \
  --max-messages 1000000 \
  --max-batches 100

Dead-letter routing and SASL/SSL

Route undecodable records to a dead-letter topic (instead of aborting the loop) with dlq_topic / --dlq-topic, and pass librdkafka settings for SASL/SSL via the consumer_conf / producer_conf dicts or repeatable --consumer-conf / --producer-conf. The count of routed records shows up as dlq_routed in StreamRunStats and the CLI output JSON.

cfg.dlq_topic = "transactions.dlq"
cfg.consumer_conf = {
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "svc",
    "sasl.password": "secret",
}
cfg.producer_conf = {"security.protocol": "SASL_SSL"}

CLI equivalent

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

Production guidance:

Key messages by entity so windows stay partition-affine.
Commit offsets only after decisions are emitted.
Use OutputDetail.DECISIONS unless you need per-rule bitmasks.
Keep rules hot-reloaded between batches, not mid-batch.