HIP-30: Event Streaming Standard. Status Draft. Hanzo's own standard — read this before implementing against it.
This proposal defines the event streaming standard for the Hanzo ecosystem. Hanzo Stream provides distributed event streaming based on Apache Kafka, serving as the backbone for real-time event ingestion, analytics pipelines, inter-service communication, and billing aggregation. All services producing or consuming asynchronous events MUST use Hanzo Stream as specified in this document.
Repository: github.com/hanzoai/stream Protocol: Kafka wire protocol (TCP 9092, TLS 9093) Production: insights-kafka on the cluster cluster (24.199.76.156)
This section explains the architectural decisions behind Hanzo Stream. Understanding the why is essential for making correct integration choices.
KRaft (Kafka Raft) is Kafka's built-in consensus protocol, available since Kafka 3.3 and production-ready since Kafka 3.6. It replaces ZooKeeper for metadata management (broker registration, topic configuration, partition leadership).
Before KRaft, running Kafka meant running two distributed systems: Kafka + ZooKeeper. ZooKeeper was a separate failure domain (session timeouts, snapshot corruption). Most Kafka outages trace back to ZooKeeper.
With KRaft, Kafka manages its own metadata. Single StatefulSet, no external dependencies. ZooKeeper is deprecated and will be removed in Kafka 4.0.
The following diagram shows how Hanzo Stream fits into the broader data flow:
┌──────────────────┐
│ Hanzo Datastore │
│ (Analytics DB) │
└────────▲─────────┘
│ consume
│
┌─────────────┐ ┌──────────────┐ ┌─────────┴─────────┐ ┌──────────────┐
│ Insights │───→│ │───→│ Consumer Groups │ │ Billing │
│ Capture │ │ │ │ │ │ Aggregator │
│ (Rust) │ │ Hanzo │ │ analytics-ingest │ └──────▲───────┘
└─────────────┘ │ Stream │ │ billing-agg │ │
│ (Kafka) │ │ audit-archive │ │ consume
┌─────────────┐ │ │ │ llm-metrics │ ┌──────┴───────┐
│ LLM Gateway │───→│ Topics: │ └───────────────────┘ │ Consumer │
│ (HIP-4) │ │ events_* │ │ Group: │
└─────────────┘ │ llm_usage │ │ billing-agg │
│ billing_* │ └──────────────┘
┌─────────────┐ │ audit_log │
│ IAM │───→│ │
│ (hanzo.id) │ └──────────────┘
└─────────────┘Data flow summary:
llm_usage events and produces billing_eventsaudit_log and writes to long-term storageAll Kafka topics MUST be registered in this section. Ad-hoc topic creation is prohibited in production.
| Topic | Partitions | Retention | Key | Producers | Consumers | |-------|-----------|-----------|-----|-----------|-----------| | events_plugin_ingestion | 16 | 7 days | team_id | Insights Capture | Hanzo Datastore Ingestion | | llm_usage | 8 | 14 days | team_id | LLM Gateway | Billing Aggregator, Analytics | | billing_events | 4 | 30 days | org_id | Billing Aggregator | Commerce, Reporting | | audit_log | 4 | 90 days | org_id | IAM, all services | Audit Archiver, Compliance | | agent_rpc_metering | 8 | 7 days | agent_id | Bot Gateway | RPC Billing (HIP-25) | | dead_letter | 1 | 30 days | source_topic | All consumers | Ops team (manual) |
<domain>_<entity>_<action>Examples: events_plugin_ingestion, llm_usage, billing_events, audit_log.
Topics MUST use snake_case. Topics MUST NOT include environment prefixes (e.g., prod_ or stg_). Environment isolation is achieved through separate Kafka clusters, not topic naming.
Partitioning determines parallelism and ordering. The key design principle: events that must be processed in order MUST share a partition key.
| Topic | Partition Key | Rationale | |-------|--------------|-----------| | events_plugin_ingestion | team_id | All events for a team arrive in order; Hanzo Datastore ingests per-team batches | | llm_usage | team_id | Billing must see all usage for a team in order to compute running totals | | billing_events | org_id | Organization-level billing aggregation | | audit_log | org_id | Audit trail must be ordered per organization | | agent_rpc_metering | agent_id | Per-agent metering must be ordered for accurate tallying |
Partition count guidelines:
max(expected_consumer_count * 2, 4)All events MUST conform to a base envelope. The envelope wraps topic-specific payloads.
{
"schema": "hanzo.stream.v1",
"id": "evt_01HQ3X7K8M2N4P5R6S7T8U9V0W",
"type": "llm.usage.completed",
"source": "llm-gateway",
"timestamp": "2025-01-15T10:30:00.000Z",
"team_id": "team_abc123",
"org_id": "org_hanzo",
"data": { }
}| Field | Type | Required | Description | |-------|------|----------|-------------| | schema | string | yes | Envelope version. Always hanzo.stream.v1 for this spec. | | id | string | yes | Globally unique event ID. ULID or UUID v7 recommended. | | type | string | yes | Dot-delimited event type (e.g., llm.usage.completed). | | source | string | yes | Producing service identifier. | | timestamp | string | yes | ISO 8601 UTC timestamp with millisecond precision. | | team_id | string | no | Team identifier (used as partition key for most topics). | | org_id | string | no | Organization identifier. | | data | object | yes | Topic-specific payload. Schema defined per event type. |
llm_usage){
"schema": "hanzo.stream.v1",
"id": "evt_01HQ3X7K8M2N4P5R6S7T8U9V0W",
"type": "llm.usage.completed",
"source": "llm-gateway",
"timestamp": "2025-01-15T10:30:00.123Z",
"team_id": "team_abc123",
"org_id": "org_hanzo",
"data": {
"request_id": "req_xyz789",
"model": "zen-72b",
"provider": "together",
"prompt_tokens": 150,
"completion_tokens": 230,
"total_tokens": 380,
"latency_ms": 1250,
"cost_usd": 0.0038,
"user_id": "user_456",
"api_key_hash": "sha256:abc...",
"cache_hit": false,
"stream": true
}
}events_plugin_ingestion){
"schema": "hanzo.stream.v1",
"id": "evt_01HQ4Y8L9N3O5Q6R7S8T9U0V1X",
"type": "insights.event.captured",
"source": "insights-capture",
"timestamp": "2025-01-15T10:31:00.456Z",
"team_id": "team_abc123",
"data": {
"event": "$pageview",
"distinct_id": "user_456",
"properties": {
"$current_url": "https://hanzo.ai/dashboard",
"$browser": "Chrome",
"$os": "macOS"
},
"ip": "203.0.113.42",
"session_id": "sess_abc123",
"sent_at": "2025-01-15T10:30:59.800Z"
}
}audit_log){
"schema": "hanzo.stream.v1",
"id": "evt_01HQ5Z9M0O4P6R7S8T9U0V1W2Y",
"type": "iam.auth.login_success",
"source": "iam",
"timestamp": "2025-01-15T10:32:00.789Z",
"org_id": "org_hanzo",
"data": {
"user_id": "user_456",
"email": "user@hanzo.ai",
"method": "oauth2",
"provider": "github",
"ip": "203.0.113.42",
"user_agent": "Mozilla/5.0...",
"application": "app-hanzo"
}
}Event schemas MUST be registered in a schema registry to enable:
schema_registry:
type: json-schema # JSON Schema (not Avro - see rationale below)
url: http://localhost:8081
compatibility: BACKWARD # New schemas must be readable by old consumersWhy JSON Schema over Avro: Our producers and consumers span multiple languages (Rust capture service, Python billing, Go IAM, TypeScript gateway). JSON Schema is language-agnostic and requires no code generation. Avro has better compression but adds build-time complexity for schema compilation. At our event sizes (< 2KB average), the compression difference is negligible.
Consumer groups enable parallel processing. Each consumer in a group reads from a disjoint set of partitions.
| Consumer Group | Topic(s) | Consumers | Processing | |---------------|----------|-----------|------------| | analytics-ingest | events_plugin_ingestion | 4 | Write to Hanzo Datastore events table | | billing-agg | llm_usage | 2 | Aggregate usage per team per hour, produce billing_events | | audit-archive | audit_log | 1 | Write to S3-compatible long-term storage | | llm-metrics | llm_usage | 2 | Compute real-time latency/throughput metrics for Prometheus | | rpc-billing | agent_rpc_metering | 2 | Aggregate agent RPC usage for settlement (HIP-25) | | dead-letter-monitor | dead_letter | 1 | Alert on failed events, PagerDuty integration |
<function>-<action>Examples: analytics-ingest, billing-agg, audit-archive. Kebab-case. MUST NOT include service version or instance identifiers.
| Category | Retention | Rationale | |----------|-----------|-----------| | Analytics events | 7 days | Sufficient for reprocessing; Hanzo Datastore is the durable store | | LLM usage | 14 days | Billing reconciliation window is 14 days | | Billing events | 30 days | Monthly billing cycle + buffer | | Audit log | 90 days | Compliance requirement; also archived to S3 | | Dead letter | 30 days | Ops investigation window |
Retention is configured per topic via retention.ms. After retention expires, segments are deleted. There is no compaction (log compaction is for changelog-style topics, not event streams).
producer:
acks: all # Wait for all in-sync replicas (durability)
retries: 3 # Retry transient failures
retry_backoff_ms: 100 # Backoff between retries
max_in_flight: 5 # Max unacknowledged batches
compression: lz4 # LZ4 for speed; zstd for better ratio if CPU-bound
batch_size: 16384 # 16KB batch size
linger_ms: 5 # Wait up to 5ms to fill batch
idempotence: true # Exactly-once per partition (requires acks=all)Why acks=all: We cannot lose billing or audit events. The latency cost of waiting for replica acknowledgment (typically < 10ms intra-cluster) is acceptable.
Why LZ4 compression: LZ4 provides ~2x compression at near-zero CPU cost. For analytics events with repetitive JSON keys, this cuts network and storage usage in half.
The Insights Rust capture service is the highest-throughput producer. It receives HTTP POST requests from frontend SDKs and writes batches to Kafka.
Browser SDK ──HTTP POST──→ Capture (Rust) ──Kafka Produce──→ events_plugin_ingestion
Port 3000 ↓
Batch: 500 events Hanzo Datastore consumer
Flush: 1 second ↓
events tableThe capture service MUST:
acks=all and idempotence enableddead_letter topic on serialization or validation failureDeployed as a Kubernetes StatefulSet on the cluster using bitnami/kafka:3.7 in KRaft combined mode (broker + controller in one process).
# Key configuration (insights-kafka StatefulSet, namespace: hanzo)
image: bitnami/kafka:3.7
replicas: 1 # Single broker, partition-ready for scale
ports: [9092 (PLAINTEXT), 9093 (TLS), 9094 (CONTROLLER)]
env:
KAFKA_CFG_PROCESS_ROLES: "broker,controller"
KAFKA_CFG_NODE_ID: "0"
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "0@localhost:9094"
KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9094"
KAFKA_CFG_LOG_RETENTION_HOURS: "168" # 7 days default
KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false" # Explicit topic creation only
resources:
requests: { memory: 1Gi, cpu: 500m }
limits: { memory: 2Gi, cpu: 1000m }
storage: 50Gi (do-block-storage PVC)Topics MUST be created explicitly. Auto-creation is disabled in production.
BROKER=localhost:9092
# Each topic: --partitions N --replication-factor 1 --config retention.ms=MS
kafka-topics.sh --bootstrap-server $BROKER --create \
--topic events_plugin_ingestion --partitions 16 --config retention.ms=604800000
kafka-topics.sh --bootstrap-server $BROKER --create \
--topic llm_usage --partitions 8 --config retention.ms=1209600000
kafka-topics.sh --bootstrap-server $BROKER --create \
--topic billing_events --partitions 4 --config retention.ms=2592000000
kafka-topics.sh --bootstrap-server $BROKER --create \
--topic audit_log --partitions 4 --config retention.ms=7776000000
kafka-topics.sh --bootstrap-server $BROKER --create \
--topic agent_rpc_metering --partitions 8 --config retention.ms=604800000
kafka-topics.sh --bootstrap-server $BROKER --create \
--topic dead_letter --partitions 1 --config retention.ms=2592000000The current deployment is a single broker. This is sufficient for our current throughput (< 50 MB/s). The scaling path is:
Scaling from 1 to 3 brokers requires:
replicas: 3KAFKA_CFG_CONTROLLER_QUORUM_VOTERS to include all three nodeskafka-reassign-partitions.shPrometheus metrics are exposed via JMX Exporter. The critical metrics:
| Metric | Alert Threshold | Meaning | |--------|----------------|---------| | kafka_messages_in_per_sec | - | Throughput: messages received per second | | kafka_bytes_in_per_sec | - | Throughput: bytes received per second | | kafka_under_replicated_partitions | > 0 for 5m | Partitions without enough replicas | | kafka_consumer_group_lag | > 100K warn, > 1M critical | Events a consumer is behind |
Consumer lag is the single most important Kafka metric. It measures how far behind a consumer group is from the latest event. If lag grows continuously, the consumer cannot keep up and events will expire before being processed.
SASL/PLAIN authentication. Each service gets a dedicated credential with scoped topic access. Passwords are injected from KMS via Kubernetes secrets.
sasl:
mechanism: PLAIN
credentials:
- { username: insights-capture, topics: [events_plugin_ingestion], ops: [WRITE] }
- { username: llm-gateway, topics: [llm_usage], ops: [WRITE] }
- { username: iam-service, topics: [audit_log], ops: [WRITE] }
- { username: analytics-ingest, topics: [events_plugin_ingestion], ops: [READ] }
- { username: billing-aggregator, topics: [llm_usage, billing_events], ops: [READ, WRITE] }Each producer and consumer MUST have minimal permissions:
| Principal | Topic | Operations | Rationale | |-----------|-------|------------|-----------| | insights-capture | events_plugin_ingestion | WRITE | Capture only produces | | llm-gateway | llm_usage | WRITE | Gateway only produces | | analytics-ingest | events_plugin_ingestion | READ | Hanzo Datastore consumer | | billing-aggregator | llm_usage | READ | Reads usage | | billing-aggregator | billing_events | WRITE | Produces billing | | audit-archiver | audit_log | READ | Archives to S3 |
No service gets blanket access to all topics. The dead_letter topic is writable by all authenticated producers (events that fail processing are redirected here).
Kafka is accessible only from within the hanzo namespace via Kubernetes NetworkPolicy. No external ingress. Services outside the cluster MUST use the LLM Gateway or Insights capture HTTP endpoints, which produce to Kafka internally.
# Check lag, then scale or reset offsets
kafka-consumer-groups.sh --bootstrap-server insights-kafka-0:9092 \
--group analytics-ingest --describe
kubectl scale deployment analytics-ingest --replicas=8 # if lag growingSingle-broker failure causes full outage. Mitigations: persistent volume survives pod restart; producers buffer ~30s in-memory; capture returns HTTP 503 (frontend SDKs retry with backoff); on restart, consumers resume from last committed offset. Multi-broker (Phase 2) with replication factor 3 eliminates single-point failure.
Copyright and related rights waived via CC0.