Server Configuration
Crabka brokers are configured from a TOML file. The fields below are the top-level keys and nested tables of that file.
| Field | Type | Required | Default | Description |
|---|---|---|---|---|
authorization | object | no | Pluggable cluster authorizer + super-user list. None ⇒ [crate::authorizer::AllowAllAuthorizer] with empty super-users (default-on-no-config behavior). When Some, the type field selects the authorizer implementation; for type = "opa", the [authorization.opa] subtable is required. | |
authorization.opa | object | no | Some iff authz_type == Opa. Required in that case; apply_to returns [FileConfigError::MissingSection] when omitted. | |
authorization.opa.allow_on_error | boolean | no | false | Security-sensitive. Permit the operation when the OPA call fails (timeout, 5xx, parse error). When true, an OPA outage authorizes every request (fail-open). Default false (fail-closed) — omitting this field denies on error, matching the upstream Open Policy Agent Kafka plugin's allow.on.error = false. |
authorization.opa.expire_after_ms | integer | no | 3600000 | Decision TTL, in milliseconds. Default 3_600_000 (1 h). |
authorization.opa.maximum_cache_size | integer | no | 50000 | LRU cache capacity, in entries. Default 50_000. |
authorization.opa.url | string | yes | OPA decision endpoint URL — must include the data-API path, e.g. http://opa:8181/v1/data/kafka/authz/allow. | |
authorization.super_users | array | no | [] | |
authorization.type | string | no | Which [crate::authorizer::Authorizer] impl to instantiate. snake_case to match the spec's type = "allow_all" | "simple" | "opa" wire shape. | |
broker_id | integer | no | ||
controller_listener_protocol | string | no | null | Controller listener security protocol. When Some(Ssl) the controller listener terminates TLS using tls_config. |
delegation_token | object | no | KIP-48: delegation-token master key + lifetime knobs. Env var CRABKA_DELEGATION_TOKEN_SECRET_KEY wins over secret_key here. When neither source provides a key, the broker disables delegation-token auth. | |
delegation_token.default_renew_period_ms | integer | no | Default renew period — the initial expiry_timestamp_ms offset at create time and the implicit renew period when RenewDelegationToken.renew_period_ms == -1. Distinct from max_lifetime_ms (the absolute ceiling). Default 24 hours. | |
delegation_token.expiry_check_interval_ms | integer | no | Background sweep cadence, ms. Default 1 hour. | |
delegation_token.max_lifetime_ms | integer | no | Hard upper bound on token lifetime, ms. Default 7 days. | |
delegation_token.secret_key | string | no | HMAC master key. Overridden by CRABKA_DELEGATION_TOKEN_SECRET_KEY when set. Bytes are wrapped in [crabka_security::SecretBytes] before reaching BrokerConfig. | |
extra_log_dirs | array | no | [] | Additional JBOD data directories (KIP-113). Maps to [crate::BrokerConfig::extra_log_dirs]. |
gssapi | object | no | SASL/GSSAPI (Kerberos) accept-path config. Broker-global — there is one [gssapi] block per broker. Relevant when a listener enables the GSSAPI mechanism. | |
gssapi.kdc | string | no | null | KDC endpoint (e.g. tcp://kdc:88) that bypasses krb5.conf discovery; falls back to krb5.conf when omitted. |
gssapi.keytab_path | string | yes | ||
gssapi.principal_to_local_rules | array | no | [] | auth_to_local rule specs, applied in order (first match wins). |
gssapi.realm | string | no | null | Default Kerberos realm, used for principals that omit their realm. |
gssapi.service_name | string | no | null | sasl.kerberos.service.name. Defaults to "kafka" when omitted. |
inter_broker_credentials | object | no | Credentials this broker uses to authenticate to peer brokers (inter-broker initiate path). Only the gssapi variant is supported. | |
inter_broker_listener_name | string | no | ||
listeners | array | no | ||
log_dir | string | no | ||
max_connections | integer | no | null | Maximum number of live broker connections across all listeners (Kafka max.connections). Connections accepted past this ceiling are closed immediately. Absent leaves the BrokerConfig default usize::MAX (unlimited), matching Kafka's Integer.MAX_VALUE. |
max_connections_per_ip | integer | no | null | Maximum number of live connections from any single client IP (Kafka max.connections.per.ip). Absent leaves the BrokerConfig default usize::MAX (unlimited). |
oauthbearer | object | no | SASL/OAUTHBEARER validator tuning. Only relevant when a listener enables the OAUTHBEARER mechanism. | |
oauthbearer.allowable_clock_skew_ms | integer | no | null | Clock-skew tolerance, in milliseconds, for exp / iat / nbf. Default 30000. |
oauthbearer.custom_claim_check | string | no | null | Optional JsonPath expression (RFC 9535, via jsonpath-rust) evaluated against the token claim set. Token is rejected when the expression yields empty/null/false. Compiled once at broker startup; malformed expressions panic with a descriptive error. |
oauthbearer.expected_audience | string | no | null | When set, the token aud claim must contain this. Signed validator only. |
oauthbearer.fallback_user_name_claim | string | no | null | Alternate claim name for principal-name fallback. |
oauthbearer.fallback_user_name_prefix | string | no | null | Prepended on fallback only. |
oauthbearer.groups_claim | string | no | null | JsonPath expression (RFC 9535) extracting groups. Compiled once at broker startup; malformed expression panics with descriptive error. |
oauthbearer.groups_claim_delimiter | string | no | null | When groups_claim resolves to a string, split on this delimiter. |
oauthbearer.idp_tls_trust | string | no | null | PEM file containing the CA certificate(s) used to verify the IdP's TLS certificate on ALL outbound HTTPS to the IdP — JWKS endpoint, introspection endpoint, and userinfo endpoint. When set, these are the only trust roots used for the outbound HTTPS (replaces the default webpki-roots — Strimzi-shaped). When unset, the broker uses reqwest's default rustls webpki-roots. |
oauthbearer.introspection_client_id | string | no | null | client_id the broker uses to authenticate (HTTP Basic Auth) against the introspection endpoint. Required when introspection_endpoint_uri is set. |
oauthbearer.introspection_client_secret_path | string | no | null | Filesystem path to a file containing the client secret the broker uses to authenticate against the introspection endpoint. Required when introspection_endpoint_uri is set. File-based (not literal) so secret material doesn't sit in the TOML; operator mounts a Secret and writes the mount path here. The file's trailing newline (if any) is stripped at config-load. |
oauthbearer.introspection_endpoint_uri | string | no | null | RFC 7662 introspection endpoint URL. When set, selects the introspection validator (mutually exclusive with jwks_endpoint_uri). |
oauthbearer.introspection_http_timeout_ms | integer | no | null | Timeout for the introspection (and userinfo) HTTP requests, in milliseconds. Default 10 000 (10 s). |
oauthbearer.jwks_endpoint_uri | string | no | null | JWKS endpoint URL. When set, tokens are validated as signed JWTs (RS256 / ES256) against the keys fetched from this URL, and the broker spawns a background refresher. When unset, the unsecured-JWS (alg:none) development validator is used. |
oauthbearer.jwks_expiry_seconds | integer | no | null | Maximum age (seconds) of the cached JWKS before validators reject tokens until the next successful refresh. Strimzi default 360 (6 minutes). Unset = no expiry check. Fails closed on prolonged IdP outage. Signed validator only. |
oauthbearer.jwks_ignore_key_use | boolean | no | null | When true, the JWKS parser keeps keys regardless of use field. Default false (filter out use=enc). Some identity providers publish signing keys with use="enc" by mistake; operators set this to true to accept them. Signed validator only. |
oauthbearer.jwks_min_refresh_pause_seconds | integer | no | null | Minimum pause (seconds) between on-demand JWKS refreshes triggered by validator signals (unknown-kid / bad-signature tokens). Defaults to 1 (Strimzi parity). Signed validator only. |
oauthbearer.jwks_refresh_interval_ms | integer | no | null | JWKS re-fetch interval, in milliseconds. Default 300000 (5 minutes). Signed validator only. |
oauthbearer.max_session_lifetime_seconds | integer | no | null | Optional ceiling on OAUTHBEARER session lifetime, in seconds. When set, the broker clamps session_lifetime_ms to min(token_exp_ms - now_ms, cap * 1000). When unset, sessions last until the token's natural exp. |
oauthbearer.principal_claim_name | string | no | null | Claim whose value becomes the principal name. Default sub. |
oauthbearer.userinfo_endpoint_uri | string | no | null | Optional OIDC userinfo endpoint URL. When set, the introspection validator calls GET userinfo after a successful introspection and merges the profile claims over the introspection claims (introspection wins for active, exp, iat, nbf, scope, client_id, sub). |
oauthbearer.valid_issuer_uri | string | no | null | When set, the token iss claim must equal this. Signed validator only. |
oauthbearer.valid_token_type | string | no | null | Optional JWT typ header check. When set, JWT-mode validators (unsecured + signed JWS) require the JWT header's typ field to equal this string. Introspection-mode skips (no JWT header). Ignored when unset. |
process | object | no | [process] section — KRaft process.roles. Absent / empty leaves the BrokerConfig default [Controller, Broker]. | |
process.roles | array | no | [] | Role strings: "controller", "broker" (case-insensitive). Empty or absent leaves the BrokerConfig default [Controller, Broker]. |
rack | string | no | null | KIP-392: this broker's rack id. Maps to BrokerConfig::rack. |
remote_storage | object | no | KIP-405: tiered-storage enablement. Setting storage_dir turns tiered storage on broker-wide and roots the local reference RemoteStorageManager there. | |
remote_storage.kafka_metadata | object | no | Opt-in to the topic-backed RemoteLogMetadataManager. When absent, the broker uses the in-memory fixture. | |
remote_storage.kafka_metadata.bootstrap | string | no | `` | host:port the manager dials to reach its own broker. May be empty; the broker derives the address from the inter-broker listener at startup. |
remote_storage.kafka_metadata.in_memory | boolean | no | false | Explicit opt-out: run the non-durable in-memory RLMM instead of the topic-backed default. Tests / single-node dev only. |
remote_storage.kafka_metadata.num_partitions | integer | no | null | Partition count for __remote_log_metadata on first creation. Defaults to 50 (Kafka's remote.log.metadata.topic.num.partitions). |
remote_storage.kafka_metadata.replication | integer | no | null | Replication factor for __remote_log_metadata on first creation. Defaults to 3 (Kafka's remote.log.metadata.topic.replication.factor). |
remote_storage.s3 | object | no | S3-compatible backend parameters. Omit to use storage_dir. | |
remote_storage.s3.access_key_id | string | no | null | Explicit access key id. Falls back to the AWS credential chain (env vars, instance profile, …) when omitted. |
remote_storage.s3.allow_http | boolean | no | false | Allow plaintext HTTP (off-by-default; required by MinIO running without TLS). |
remote_storage.s3.bucket | string | yes | S3 bucket name. | |
remote_storage.s3.endpoint | string | no | null | Optional custom endpoint URL (e.g. MinIO or Cloudflare R2). |
remote_storage.s3.multipart_chunk_size | integer | no | null | Optional override of the per-part multipart chunk size (bytes). When None, [crabka_remote_storage::DEFAULT_MULTIPART_CHUNK_SIZE] applies. AWS requires parts ≥ 5 MiB except the last; MinIO tolerates smaller values. |
remote_storage.s3.multipart_threshold | integer | no | null | Optional override of the multipart-upload threshold (bytes). When None, [crabka_remote_storage::DEFAULT_MULTIPART_THRESHOLD] applies. Operators typically leave this alone; lower it to force multipart on smaller segments for testing. |
remote_storage.s3.prefix | string | no | null | Optional key prefix inside the bucket (lets multiple clusters share a bucket). |
remote_storage.s3.region | string | yes | AWS region. Required even for non-AWS endpoints (use any value). | |
remote_storage.s3.secret_access_key | string | no | null | Explicit secret access key. Falls back to the AWS credential chain when omitted. |
remote_storage.storage_dir | string | no | Root directory for the local LocalTieredStorage backend. | |
replica_selector | string | no | null | KIP-392: replica selector name ("leader" | "rack-aware"). Maps to BrokerConfig::replica_selector. |
server_properties | object | no | {} | |
super_users | array | no | Principals that are unconditionally authorized for all operations, including KIP-48 delegation-token act-as. The operator emits super_users = ["ANONYMOUS"] when Kafka.spec.delegationToken is set so its PLAINTEXT inter-broker reconcile loop can mint per-KafkaUser tokens. None and Some(empty) are equivalent — both leave BrokerConfig.super_users empty. | |
tls_config | object | no | TLS material for the controller listener (and any listener whose protocol is TLS-bearing). | |
tls_config.cert_path | string | yes | ||
tls_config.client_auth | string | no | ||
tls_config.client_ca_path | string | no | null | |
tls_config.key_path | string | yes |