# Kafka Conventions (cex-core + cex-identity)
The single source of truth for how every platform service configures Kafka.
Enforced in code by `road_runner_common::kafka` (Rust) and
`new-metadata/src/infrastructure/kafka/kafka.config.ts` (Node).
## 1. Config layering — where each setting lives
| **ENV** | `KAFKA_BROKERS`, `KAFKA_SECURITY_PROTOCOL`, `KAFKA_SASL_{MECHANISM,USERNAME,PASSWORD}`, `KAFKA_SSL_CA_PEM`, ops overrides (`KAFKA_SESSION_TIMEOUT_MS`) | SRE / deployment |
| **CODE** | topic names, `group.id`, `client.id`, delivery semantics (acks, idempotence, isolation, offset reset) | developers |
| **IaC** | topic lifecycle: partitions, replication factor, `min.insync.replicas`, `cleanup.policy`, retention | platform / broker chart |
**Rule of thumb:** changes per environment or is secret → ENV. Part of the app
contract or correctness → CODE. Topic existence/sizing → IaC.
## 2. Naming
- **Topic:** `<publisher>.<event>` — e.g. `cex_policy.events`,
`cex_auth.admin_audit_events`, `cex_iam.authz_snapshot`,
`keycloak.user-events`. Each **publisher owns its topic-name constants**;
consumers pin the same canonical string in their own constants.
- **`group.id`:** `<service>` or `<service>.<purpose>` (`cex-history`,
`cex-metadata.policy-events`). Owned in code, never env.
- **`client.id`:** `<service>`.
- **DLQ:** `<topic>.dlq`.
- **Metrics** (labels `topic`, `group`): `kafka_producer_messages_total`,
`kafka_producer_errors_total`, `kafka_consumer_messages_total`,
`kafka_consumer_errors_total`, `kafka_consumer_process_duration_seconds`,
`kafka_consumer_lag_seconds`. (Rust: constants in `road_runner_common::kafka::metrics`.)
## 3. Delivery semantics (fixed in code)
- **Producers:** `acks=all` + `enable.idempotence=true` + `compression=lz4`.
Durable, no duplicates on retry. Requires broker `min.insync.replicas>=2`.
- **Consumers:** `enable.auto.commit=false` (commit after processing),
`isolation.level=read_committed`, explicit `auto.offset.reset`
(`earliest` for audit/state rebuild, `latest` for snapshot/live).
## 4. Shared config — do not hand-roll `ClientConfig`
**Rust** (`Cargo.toml`: `road-runner-common = { version = "0.7.1", features = [..., "kafka"] }`).
The module returns plain `(key, value)` settings — never a version-specific
`rdkafka::ClientConfig` — so it works with any rdkafka version a service pins
(the platform spans 0.36–0.39). Apply them to your own config:
```rust
use road_runner_common::kafka::{KafkaSettings, consumer_settings, producer_settings, OffsetReset};
let settings = KafkaSettings::from_env();
let mut pcfg = rdkafka::ClientConfig::new();
for (k, v) in producer_settings(&settings) { pcfg.set(k, v); }
let producer: FutureProducer = pcfg.set("client.id", "cex-policy").create()?;
let mut ccfg = rdkafka::ClientConfig::new();
for (k, v) in consumer_settings(&settings, "cex-history", OffsetReset::Earliest) { ccfg.set(k, v); }
let consumer: StreamConsumer = ccfg.set("client.id", "cex-history").create()?;
```
`KafkaSettings::security_settings()` replaces every copied `apply_kafka_security`.
**Node** (`kafka.config.ts`): `kafkaClientConfig()` (clientId + SASL/SSL from env),
`producerConfig()` (idempotent, no auto-create), `consumerGroupId('<purpose>')`.
## 5. Topic lifecycle — IaC, not the app
Applications must **never** auto-create topics. The broker is **Bitnami Kafka**
(not Strimzi), so topics are declared via the chart's `provisioning.topics` (a
provisioning Job), and `min.insync.replicas=2` is set in the broker config to
pair with `acks=all`. Snapshot topics (`*.authz-snapshot`) use
`cleanup.policy=compact`, `retention.ms=-1`, `partitions=1`.
> The plan originally named Strimzi `KafkaTopic` CRDs; the running cluster is
> Bitnami Kafka, so the equivalent Bitnami `provisioning.topics` mechanism is used
> to reach the same declarative outcome.
## 6. Topic ownership (publisher → consumers)
| `cex_engine.order.acknowledged` / `trade.executed` / `order.rejected` | matching engine (external) | cex-history, cex-compliance, cex-stats, cex-balance |
| `cex_engine.new_order` / `market_updated` / `new_trade` | matching engine (external) | (broadcast plane) |
| `cex_policy.events` | cex-policy | identity-account, cex-iam, cex-metadata |
| `cex_iam.authz_snapshot` (compact) | cex-iam | every embedded-PDP service (via road-runner authz-kafka) |
| `cex_auth.admin_audit_events` | identity-gateway | cex-audit |
| `keycloak.user-events` / `admin-events` | keycloak SPI (vendored, env-only) | identity-account |
| `apikey.usage` | identity-account | (analytics) |
## 7. Exceptions
- **Keycloak** (`user-keycloak` Java SPI): the vendored `keycloak-kafka` provider
is configured entirely via env (`KAFKA_BOOTSTRAP_SERVERS`, `KAFKA_TOPIC`, …) — it
cannot hold code defaults. Documented exception; its topics still get IaC entries.
- **cex-socket** (C++): uses Redis pub/sub, **not Kafka**. No Kafka config.