Server Configuration

Crabka brokers are configured from a TOML file. Each section below maps to a top-level key or [table] of that file.

General

FieldTypeRequiredDefaultDescription
auto_joinbooleannonullEnable automatic dynamic controller enrollment.
bootstrap_serversarrayno[]KIP-853 controller discovery endpoints. Hosts remain unresolved so DNS names can be refreshed on each retry.
broker_idintegerno
controller_election_timeoutstringnonullController election timeout. Absent leaves the BrokerConfig default intact.
controller_heartbeat_intervalstringnonullController heartbeat interval. Absent leaves the BrokerConfig default intact.
controller_listener_protocolstringnonullController listener security protocol. When Some(Ssl) the controller listener terminates TLS using tls_config.
controller_quorum_votersarrayno[]KIP-595 static controller quorum voter set. Each entry is <node_id>@<host>:<port> pointing at a broker's controller listener (port 9093). At apply time each entry is parsed (NOT DNS-resolved) and its <host>:<port> is carried verbatim into BrokerConfig::controller_quorum_voters. The inter-broker dialer re-resolves the host on every (re)connect (TcpStream::connect), so a peer that restarts on a new pod IP (a StatefulSet pod keeps its stable DNS name but gets a fresh A record) is reached again without restarting this broker — pre-resolving here would freeze the peer's boot-time IP and strand a rejoining voter. Empty leaves the single self-voter the binary seeds (standalone).
controller_server_namestringnonullTLS server name (SNI) presented when dialing a PEER's controller listener for the KIP-595 quorum. The operator renders the shared headless-Service FQDN here — a SAN on every broker's serving cert — so mTLS validation succeeds no matter which peer (resolved to a pod IP) is dialed. Absent falls back to "localhost". Maps to [crate::BrokerConfig::controller_server_name].
extra_log_dirsarrayno[]Additional JBOD data directories (KIP-113). Maps to [crate::BrokerConfig::extra_log_dirs].
heartbeat_intervalstringnonullHow often this broker sends BrokerHeartbeat to the controller leader. Absent leaves the BrokerConfig default intact.
heartbeat_timeoutstringnonullController-side session timeout for broker heartbeats. Absent leaves the BrokerConfig default intact.
inter_broker_credentialsobjectnoCredentials this broker uses to authenticate to peer brokers (inter-broker initiate path). Only the gssapi variant is supported.
inter_broker_listener_namestringno
listenersarrayno
log_dirstringno
max_connectionsintegernonullMaximum 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_ipintegernonullMaximum number of live connections from any single client IP (Kafka max.connections.per.ip). Absent leaves the BrokerConfig default usize::MAX (unlimited).
rackstringnonullKIP-392: this broker's rack id. Maps to BrokerConfig::rack.
replica_lag_time_maxstringnonullMaximum follower lag before the leader proposes ISR shrink. Absent leaves the BrokerConfig default intact.
replica_selectorstringnonullKIP-392: replica selector name ("leader" | "rack-aware"). Maps to BrokerConfig::replica_selector.
server_propertiesobjectno{}
super_usersarraynoPrincipals 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.

audit

FedRAMP 20x MLA audit subsystem configuration. Absent → secure default (enabled, standard internal topic name).

FieldTypeRequiredDefaultDescription
checkpointobjectnoCheckpoint emission cadence. None → use defaults.
checkpoint.every_nintegerno1000
checkpoint.every_secsintegerno60
enabledbooleannotrueWhether the audit subsystem is active.
signingobjectnoEd25519 checkpoint signing key. None → chaining only, no checkpoints.
signing.key_idstringyes
signing.key_pathstringyes
spoolobjectnoDurable spool for the AU-5 degraded path. None → use defaults.
spool.dirstringnoaudit-spool
spool.max_bytesintegerno1073741824
topicstringno__crabka_auditInternal topic name for audit records.

authorization

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.

FieldTypeRequiredDefaultDescription
opaobjectnoSome iff authz_type == Opa. Required in that case; apply_to returns [FileConfigError::MissingSection] when omitted.
opa.allow_on_errorbooleannofalseSecurity-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.
opa.expire_after_msintegerno3600000Decision TTL, in milliseconds. Default 3_600_000 (1 h).
opa.maximum_cache_sizeintegerno50000LRU cache capacity, in entries. Default 50_000.
opa.urlstringyesOPA decision endpoint URL — must include the data-API path, e.g. http://opa:8181/v1/data/kafka/authz/allow.
super_usersarrayno[]
typestringnoWhich [crate::authorizer::Authorizer] impl to instantiate. snake_case to match the spec's type = "allow_all" | "simple" | "opa" wire shape.

delegation_token

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.

FieldTypeRequiredDefaultDescription
default_renew_period_msintegernoDefault 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.
expiry_check_interval_msintegernoBackground sweep cadence, ms. Default 1 hour.
max_lifetime_msintegernoHard upper bound on token lifetime, ms. Default 7 days.
secret_keystringnoHMAC master key. Overridden by CRABKA_DELEGATION_TOKEN_SECRET_KEY when set. Bytes are wrapped in [crabka_security::SecretBytes] before reaching BrokerConfig.

gssapi

SASL/GSSAPI (Kerberos) accept-path config. Broker-global — there is one [gssapi] block per broker. Relevant when a listener enables the GSSAPI mechanism.

FieldTypeRequiredDefaultDescription
kdcstringnonullKDC endpoint (e.g. tcp://kdc:88) that bypasses krb5.conf discovery; falls back to krb5.conf when omitted.
keytab_pathstringyes
max_time_skewstringnonullMaximum tolerated difference between client and broker clocks.
principal_to_local_rulesarrayno[]auth_to_local rule specs, applied in order (first match wins).
realmstringnonullDefault Kerberos realm, used for principals that omit their realm.
service_namestringnonullsasl.kerberos.service.name. Defaults to "kafka" when omitted.

oauthbearer

SASL/OAUTHBEARER validator tuning. Only relevant when a listener enables the OAUTHBEARER mechanism.

FieldTypeRequiredDefaultDescription
allowable_clock_skew_msintegernonullClock-skew tolerance, in milliseconds, for exp / iat / nbf. Default 30000.
custom_claim_checkstringnonullOptional 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.
expected_audiencestringnonullWhen set, the token aud claim must contain this. Signed validator only.
fallback_user_name_claimstringnonullAlternate claim name for principal-name fallback.
fallback_user_name_prefixstringnonullPrepended on fallback only.
groups_claimstringnonullJsonPath expression (RFC 9535) extracting groups. Compiled once at broker startup; malformed expression panics with descriptive error.
groups_claim_delimiterstringnonullWhen groups_claim resolves to a string, split on this delimiter.
idp_tls_truststringnonullPEM 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.
introspection_client_idstringnonullclient_id the broker uses to authenticate (HTTP Basic Auth) against the introspection endpoint. Required when introspection_endpoint_uri is set.
introspection_client_secret_pathstringnonullFilesystem 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.
introspection_endpoint_uristringnonullRFC 7662 introspection endpoint URL. When set, selects the introspection validator (mutually exclusive with jwks_endpoint_uri).
introspection_http_timeout_msintegernonullTimeout for the introspection (and userinfo) HTTP requests, in milliseconds. Default 10 000 (10 s).
jwks_endpoint_uristringnonullJWKS 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.
jwks_expiry_secondsintegernonullMaximum 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.
jwks_ignore_key_usebooleannonullWhen 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.
jwks_min_refresh_pause_secondsintegernonullMinimum pause (seconds) between on-demand JWKS refreshes triggered by validator signals (unknown-kid / bad-signature tokens). Defaults to 1 (Strimzi parity). Signed validator only.
jwks_refresh_interval_msintegernonullJWKS re-fetch interval, in milliseconds. Default 300000 (5 minutes). Signed validator only.
max_session_lifetime_secondsintegernonullOptional 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.
principal_claim_namestringnonullClaim whose value becomes the principal name. Default sub.
userinfo_endpoint_uristringnonullOptional 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).
valid_issuer_uristringnonullWhen set, the token iss claim must equal this. Signed validator only.
valid_token_typestringnonullOptional 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

[process] section — KRaft process.roles. Absent / empty leaves the BrokerConfig default [Controller, Broker].

FieldTypeRequiredDefaultDescription
rolesarrayno[]Role strings: "controller", "broker" (case-insensitive). Empty or absent leaves the BrokerConfig default [Controller, Broker].

remote_storage

KIP-405: tiered-storage enablement. Setting storage_dir turns tiered storage on broker-wide and roots the local reference RemoteStorageManager there.

FieldTypeRequiredDefaultDescription
gcsobjectnoNative Google Cloud Storage backend parameters. Omit to use storage_dir or [remote_storage.s3].
gcs.allow_httpbooleannofalseAllow plaintext HTTP (off-by-default; required by emulators running without TLS).
gcs.application_credentials_pathstringnonullPath to an Application Default Credentials JSON file. Omit (along with the other credential fields) to use Workload Identity / ADC.
gcs.bucketstringyesGCS bucket name.
gcs.endpointstringnonullOptional custom GCS API base URL (for emulators / fakes).
gcs.multipart_chunk_sizeintegernonullOptional override of the per-part multipart chunk size (bytes). When None, [crabka_remote_storage::DEFAULT_MULTIPART_CHUNK_SIZE] applies.
gcs.multipart_thresholdintegernonullOptional 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.
gcs.prefixstringnonullOptional key prefix inside the bucket (lets multiple clusters share a bucket).
gcs.service_account_keystringnonullInline service-account JSON key. Omit (along with the other credential fields) to use Workload Identity / ADC.
gcs.service_account_pathstringnonullPath to a service-account JSON key file. Omit (along with the other credential fields) to use Workload Identity / ADC.
kafka_metadataobjectnoOpt-in to the topic-backed RemoteLogMetadataManager. When absent, the broker uses the in-memory fixture.
kafka_metadata.bootstrapstringno``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.
kafka_metadata.event_queue_capacityintegernonullCapacity of the shared metadata-event delivery queue.
kafka_metadata.fetch_max_bytesstringnonullMaximum bytes returned by each per-partition metadata fetch.
kafka_metadata.fetch_max_waitstringnonullMaximum wait for each per-partition metadata fetch.
kafka_metadata.fetch_retry_backoffstringnonullBackoff after a failed metadata fetch.
kafka_metadata.in_memorybooleannofalseExplicit opt-out: run the non-durable in-memory RLMM instead of the topic-backed default. Tests / single-node dev only.
kafka_metadata.num_partitionsintegernonullPartition count for __remote_log_metadata on first creation. Defaults to 50 (Kafka's remote.log.metadata.topic.num.partitions).
kafka_metadata.replicationintegernonullReplication factor for __remote_log_metadata on first creation. Defaults to 3 (Kafka's remote.log.metadata.topic.replication.factor).
kafka_metadata.snapshot_intervalstringnonullRLMM cache snapshot cadence.
kafka_metadata.topic_create_timeoutstringnonullTimeout for provisioning each internal metadata topic.
s3objectnoS3-compatible backend parameters. Omit to use storage_dir.
s3.access_key_idstringnonullExplicit access key id. Falls back to the AWS credential chain (env vars, instance profile, …) when omitted.
s3.allow_httpbooleannofalseAllow plaintext HTTP (off-by-default; required by MinIO running without TLS).
s3.bucketstringyesS3 bucket name.
s3.endpointstringnonullOptional custom endpoint URL (e.g. MinIO or Cloudflare R2).
s3.multipart_chunk_sizeintegernonullOptional 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.
s3.multipart_thresholdintegernonullOptional 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.
s3.prefixstringnonullOptional key prefix inside the bucket (lets multiple clusters share a bucket).
s3.regionstringyesAWS region. Required even for non-AWS endpoints (use any value).
s3.secret_access_keystringnonullExplicit secret access key. Falls back to the AWS credential chain when omitted.
storage_dirstringnoRoot directory for the local LocalTieredStorage backend.

runtime

Operational runtime policy. Present values replace the current broker value; absent values retain it.

FieldTypeRequiredDefaultDescription
acl_max_principalstringnonull
acl_max_resource_namestringnonull
audit_event_queue_capacityintegerno
audit_partition_wait_timeoutstringnonull
audit_spool_replay_intervalstringnonull
audit_stats_poll_intervalstringnonull
audit_tail_read_maxstringnonull
audit_tail_window_offsetsintegerno
auto_join_retry_backoffstringnonull
auto_join_voter_request_timeoutstringnonull
classic_group_initial_rebalance_delaystringnonull
cleaner_intervalstringnonull
client_metrics_default_intervalstringnonull
client_metrics_eviction_tickstringnonull
client_metrics_otlp_queue_capacityintegerno
client_metrics_prom_snapshot_ttlstringnonull
client_metrics_stale_floorstringnonull
client_metrics_stale_push_intervalsintegerno
client_metrics_telemetry_maxstringnonull
connection_creation_throttle_maxstringnonull
consumer_group_heartbeat_intervalstringnonull
consumer_group_max_heartbeat_intervalstringnonull
consumer_group_max_session_timeoutstringnonull
consumer_group_max_sizeintegerno
consumer_group_min_heartbeat_intervalstringnonull
consumer_group_min_session_timeoutstringnonull
consumer_group_session_timeoutstringnonull
controlled_shutdown_drain_timeoutstringnonull
controller_election_timeoutstringnonull
controller_fetch_miss_limitintegerno
controller_heartbeat_intervalstringnonull
coordinator_actor_mailbox_capacityintegerno
coordinator_session_expiry_tickstringnonull
coordinator_shutdown_ack_timeoutstringnonull
default_min_insync_replicasintegerno
delegation_token_default_renew_periodstringnonull
delegation_token_expiry_check_intervalstringnonull
delegation_token_max_lifetimestringnonull
diskless_wal_local_replica_countintegerno
future_log_move_read_chunkstringnonull
future_log_move_retry_backoffstringnonull
gauge_poll_intervalstringnonull
heartbeat_intervalstringnonull
heartbeat_timeoutstringnonull
inter_broker_server_namestringno
isr_scan_intervalstringnonull
leader_imbalance_check_intervalstringnonull
leader_imbalance_per_brokerstringnonull
liveness_tick_intervalstringnonull
log_read_buffer_capstringnonull
log_timestamp_scan_windowstringnonull
max_connectionsintegerno
max_connections_per_ipintegerno
max_incremental_fetch_session_cache_slotsintegerno
max_produce_groupintegerno
metadata_max_between_snapshotsstringnonull
metadata_max_snapshot_intervalstringnonull
metadata_raft_command_queue_capacityintegerno
metadata_raft_fetch_maxstringnonull
metadata_snapshot_fetch_maxstringnonull
metadata_snapshot_interval_recordsintegerno
oauth_jwks_http_timeoutstringnonull
observer_fetch_maxstringnonull
observer_lag_boundintegerno
observer_poll_intervalstringnonull
offsets_topic_metadata_wait_timeoutstringnonull
opa_http_timeoutstringnonull
operator_recovery_deadlinestringnonull
partition_disk_scan_intervalstringnonull
partition_writer_queue_depthintegerno
producer_id_expirationstringnonull
producer_id_expiration_scan_intervalstringnonull
quota_throttle_maxstringnonull
record_decompression_max_ratiostringnonull
record_decompression_output_ceilingstringnonull
record_decompression_output_floorstringnonull
remote_log_manager_intervalstringnonull
replica_lag_time_maxstringnonull
replication_epoch_fence_backoffstringnonull
replication_fetch_maxstringnonull
replication_fetch_max_waitstringnonull
replication_fetch_minstringnonull
replication_reconnect_delay_capstringnonull
replication_reconnect_initial_delaystringnonull
replication_send_error_backoffstringnonull
replication_throttle_exhausted_backoffstringnonull
replication_unexpected_error_backoffstringnonull
replication_unknown_topic_retry_delaystringnonull
rlmm_bootstrap_backoff_initialstringnonull
rlmm_bootstrap_backoff_maxstringnonull
rlmm_reconcile_tickstringnonull
self_registration_backoff_maxstringnonull
self_registration_backoff_minstringnonull
self_registration_max_attemptsintegerno
sendfile_minstringnonull
share_group_enablebooleanno
share_group_heartbeat_intervalstringnonull
share_group_isolation_levelstringno
share_group_max_delivery_attemptsintegerno
share_group_max_inflight_recordsintegerno
share_group_record_lock_durationstringnonull
share_group_session_timeoutstringnonull
share_recovery_read_maxstringnonull
share_session_cache_max_when_unlimitedintegerno
share_state_num_partitionsintegerno
share_state_replication_factorintegerno
socket_receive_bufferstringnonull
socket_request_maxstringnonull
socket_send_bufferstringnonull
startup_leader_wait_timeoutstringnonull
streams_group_acceptable_recovery_lagintegerno
streams_group_assignorstringno
streams_group_heartbeat_intervalstringnonull
streams_group_num_standby_replicasintegerno
streams_group_num_warmup_replicasintegerno
streams_group_session_timeoutstringnonull
streams_group_task_offset_intervalstringnonull
streams_internal_topic_replication_factorintegerno
sync_group_follower_waitstringnonull
telemetry_decompressed_output_ceilingstringnonull
telemetry_decompressed_output_floorstringnonull
telemetry_max_decompression_ratiostringnonull
tls_reload_intervalstringnonull
transaction_max_timeoutstringnonull
transaction_min_timeoutstringnonull
transaction_recovery_read_maxstringnonull
transaction_state_num_partitionsintegerno
transaction_state_replication_factorintegerno
txn_abort_cleanup_intervalstringnonull
unclean_recovery_aggressive_deadlinestringnonull
unclean_recovery_balanced_deadlinestringnonull
unclean_recovery_queue_capacityintegerno

tls_config

TLS material for the controller listener (and any listener whose protocol is TLS-bearing).

FieldTypeRequiredDefaultDescription
cert_pathstringyes
client_authstringno
client_ca_pathstringnonull
key_pathstringyes
trust_roots_pathstringnonullPEM file of CA(s) this broker trusts when validating a PEER's server cert as an outbound inter-broker / controller-quorum dialer. The operator renders the cluster CA here so KIP-595 controller peers can mutually authenticate over the controller listener. Maps to [crabka_security::TlsConfig::trust_roots_path].