Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ All notable changes to this project will be documented in this file.

- Support floating tags for product images via the new `spec.image.stackableVersionPolicy` field
([#1021]).
- Support Kerberos (GSSAPI) authentication on KRaft controllers, covering both
broker-to-controller and controller-to-controller (Raft) traffic ([#1024]).

### Changed

Expand All @@ -21,9 +23,6 @@ All notable changes to this project will be documented in this file.
that adds the new controller to the voter list.
On termination, a new `preStop` hook on the controller container (`kafka`) removes the pod from
the voter list before shutdown.
The property `controller.quorum.bootstrap.servers` now contains the headless service names
of all controller role groups instead of individual peer host names. This prevents the
restart controller from restarting all pods in the quorum when a new one is added/deleted.
The controller `StatefulSet` is now scaled using `OrderedReady` instead of the `Parallel` strategy
to ensure only one voter is added/removed at a time and thus keep the quorum healthy ([#1010]).
- Internal operator refactoring: introduce a build() step in the reconciler that
Expand Down Expand Up @@ -76,6 +75,7 @@ All notable changes to this project will be documented in this file.
[#1014]: https://github.com/stackabletech/kafka-operator/pull/1014
[#1017]: https://github.com/stackabletech/kafka-operator/pull/1017
[#1021]: https://github.com/stackabletech/kafka-operator/pull/1021
[#1024]: https://github.com/stackabletech/kafka-operator/pull/1024

## [26.7.0] - 2026-07-21

Expand Down
27 changes: 24 additions & 3 deletions docs/modules/kafka/pages/usage-guide/kraft-controller.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -92,13 +92,34 @@ KRaft mode requires major configuration changes compared to ZooKeeper:
with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself.
Every other controller formats with `--no-initial-controllers` and joins purely through the sidecar's `add-controller` call.
Brokers always format with `--no-initial-controllers` too; they are never voters.
* `controller.quorum.bootstrap.servers` points at each controller role group's own headless Service DNS name, not
individual pod addresses.
* `controller.quorum.bootstrap.servers` lists the individual pod FQDNs of all controllers.
This is required by Kerberos -- a GSSAPI client derives the service principal from the hostname it dials, and the
CONTROLLER listener can only offer the pod's own principal -- but it means the property changes whenever a
controller role group is scaled, so the controller pods are rolled by the restart controller on scale.

== Kerberos

Apache Kafka KRaft controllers support Kerberos (GSSAPI) authentication.
Reference an `AuthenticationClass` with the Kerberos provider as described in
xref:usage-guide/security.adoc[Security].

When Kerberos is enabled:

* The `CONTROLLER` listener uses `SASL_SSL` instead of `SSL`, with `GSSAPI` as the mechanism
(`sasl.mechanism.controller.protocol`). This covers both broker-to-controller traffic and
the controller-to-controller (Raft) traffic between quorum members.
* Controller keytabs are *pod-scoped*, unlike broker keytabs, which are scoped to their
listener volumes. Controllers are reachable only under their own StatefulSet pod DNS name,
which is therefore the principal in their keytab.
* Each pod's `jaas.properties` gains a `controller.KafkaServer` login context. Unlike the
broker-side contexts it does not set `isInitiator=false`, because a controller must be able
to initiate GSSAPI connections to its peers, not only accept them.
* The `quorum-manager` sidecar and the `preStop` hook authenticate to the CONTROLLER listener
as the controller's own pod principal.

== Known Issues

* Automatic migration from Apache ZooKeeper to KRaft is not supported.
* Kerberos is currently not supported for KRaft in all versions.
* The quorum is created once by the controller with the lowest `node.id` using `--standalone`. If this controller
loses it's PVC, a new conflicting quorum is created on restart.
* A Controller that loses its persistent volume is not re-admitted to the voter set automatically, because it
Expand Down
54 changes: 48 additions & 6 deletions docs/modules/kafka/pages/usage-guide/security.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -162,13 +162,55 @@ NOTE: When Kerberos is enabled it is also required to enable TLS for maximum sec

==== Clients

In order to keep client configuration as uncluttered as possible, each kerberized Kafka broker has two principals: one for the broker itself and one for the bootstrap service.
The client can connect to the bootstrap service, which returns the broker quorum for use in subsequent operations.
This is transparent as each connection dynamically uses the relevant principal (broker or bootstrap).
In order for this to work, it is necessary for kerberized clusters to define an extra Kafka listener for the bootstrap with a corresponding service (and port).
The bootstrap address is written to the discovery ConfigMap, using the Stackable bootstrap listener with the port being 9095 (secure) for kerberized clusters, and 9092 (non-secure) or 9093 (secure) for non-kerberized ones.
===== Why each broker has two principals

NOTE: Port 9094 is reserved for non-secure kerberized connections which is not currently implemented.
A GSSAPI client derives the service principal it asks the KDC for from the *hostname it connects to*.
Connecting to a Kafka cluster takes two hops over two different addresses: a client first contacts a bootstrap address, receives the cluster metadata, and then connects to individual brokers directly.
Each of those addresses therefore needs its own principal.

Every broker consequently gets two principals, `kafka/<bootstrap-address>` and `kafka/<broker-address>`, both provisioned into its keytab by the Secret Operator.
The broker's JAAS configuration declares both, as the `bootstrap.KafkaServer` and `client.KafkaServer` login contexts.
No client-side configuration is needed to switch between them: whichever address a client dials, the broker already holds the matching principal.

To make this work, a kerberized cluster exposes an additional Kafka listener, container port and `Listener` port for the bootstrap address.
These exist only when Kerberos is enabled; without it, clients reach brokers over the client listener alone.

===== Bootstrap address and ports

The bootstrap address is published in the xref:reference/discovery.adoc[discovery ConfigMap] under the `KAFKA` key, read from the ingress addresses of the Stackable bootstrap `Listener`.
The port depends on whether Kerberos and TLS are enabled:

[cols="1,1,1"]
|===
| Cluster | Port name | Port

| Kerberos (always TLS)
| `bootstrap`
| 9095

| TLS, no Kerberos
| `kafka-tls`
| 9093

| No TLS, no Kerberos
| `kafka`
| 9092
|===

NOTE: Kerberos requires TLS, so a kerberized cluster always uses port 9095 for bootstrapping.

===== Client configuration

The discovery ConfigMap's `client.properties` carries the properties needed to *reach* the cluster:

- `security.protocol`
- `sasl.mechanism`
- `sasl.kerberos.service.name`
- and the truststore settings

But the operator adds *no login configuration* to 'client.properties` because supplying credentials is the client's responsibility.
The operator cannot do it: these clients run outside the Kafka Pods, so they have neither the Pods' keytabs nor their principals.
Add your own `sasl.jaas.config`, or point the JVM at a JAAS file with `java.security.auth.login.config`, naming the principal and keytab the client should authenticate with.

== [[authorization]]Authorization

Expand Down
9 changes: 4 additions & 5 deletions docs/modules/kafka/partials/supported-versions.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,11 @@
// This is a separate file, since it is used by both the direct Kafka documentation, and the overarching
// Stackable Platform documentation.

* 4.2.1 (experimental, deprecated) - Requires KRaft, please read on the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide].
* 4.1.1 (experimental, deprecated) - Requires KRaft, please read on the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide].
* 4.2.1
* 4.1.1
* 3.9.2 (LTS)
* 3.9.1 (deprecated)

Support for clusters running in Kraft mode (which includes Apache Kafka >= 4.x) is experimental due to the following known issues:
Starting with Apache Kafka >=4.x, the quorum manager has been changed from Apache Zookeeper to the built in KRaft.

* Kerberos authentication is not tested yet.
* Service exposition is not definitive.
If you are running an older Kafka version and plan to upgrade, head on to the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide] for details on how to do so.
107 changes: 96 additions & 11 deletions rust/operator-binary/src/controller/build/command.rs
Original file line number Diff line number Diff line change
Expand Up @@ -142,13 +142,15 @@ fn controller_quorum_format_flag(controller_descriptors: &[KafkaPodDescriptor])
}

pub fn controller_kafka_container_command(
kafka_security: &ValidatedKafkaSecurity,
controller_descriptors: Vec<KafkaPodDescriptor>,
) -> String {
formatdoc! {"
{COMMON_BASH_TRAP_FUNCTIONS}
{remove_vector_shutdown_file_command}
prepare_signal_handlers
containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
{set_realm_env}

{derive_pod_index}
{export_replica_id}
Expand All @@ -157,6 +159,12 @@ pub fn controller_kafka_container_command(

config-utils template /tmp/{properties_file}

cp {config_dir}/{jaas_file} /tmp/{jaas_file}
config-utils template /tmp/{jaas_file}

cp {admin_client_source} {admin_client_config}
config-utils template {admin_client_config}

{quorum_format_flag}
bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted \"$FORMAT_QUORUM_FLAG\"
bin/kafka-server-start.sh /tmp/{properties_file} &
Expand All @@ -165,18 +173,30 @@ pub fn controller_kafka_container_command(
{create_vector_shutdown_file_command}
",
remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
// Mirrors `broker_kafka_container_commands`: empty when Kerberos is disabled.
set_realm_env = match kafka_security.has_kerberos_enabled() {
true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})"),
false => "".to_string(),
},
derive_pod_index = DERIVE_POD_INDEX,
export_replica_id = EXPORT_REPLICA_ID,
config_dir = STACKABLE_CONFIG_DIR,
properties_file = ConfigFileName::ControllerProperties,
jaas_file = ConfigFileName::Jaas,
admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH,
admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH,
quorum_format_flag = controller_quorum_format_flag(&controller_descriptors),
create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
}
}

const KAFKA_METADATA_QUORUM_BINARY: &str = "/stackable/kafka/bin/kafka-metadata-quorum.sh";

const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/stackable/config/admin-client.properties";
/// The rendered admin-client config. The raw ConfigMap file is copied here and passed through
/// `config-utils template` first, because under Kerberos its `sasl.jaas.config` carries
/// `${env:...}` placeholders (see `controller_admin_client_properties`).
const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/tmp/admin-client.properties";
const ADMIN_CLIENT_PROPERTIES_SOURCE_PATH: &str = "/stackable/config/admin-client.properties";

/// The merged config used only for `add-controller` (self-registration).
///
Expand All @@ -200,11 +220,6 @@ const CLI_CALL_KILL_AFTER_SECONDS: u32 = 5;
/// Shell snippet setting `$BOOTSTRAP_SERVERS` by extracting
/// `controller.quorum.bootstrap.servers` from the static, un-rendered `controller.properties`
/// ConfigMap file.
///
/// Reading this at runtime, rather than baking the peer list into this script as a Rust
/// literal, keeps both sidecar scripts' content — and therefore the controller pod
/// template — identical across changes to an existing controller role group's *replica
/// count*.
fn extract_bootstrap_servers_command() -> String {
format!(
r#"BOOTSTRAP_SERVERS=$(grep '^controller.quorum.bootstrap.servers=' {config_dir}/{controller_properties_file} | cut -d= -f2- | sed 's/\\:/:/g')"#,
Expand All @@ -230,6 +245,12 @@ const CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT: &str =
/// The sidecar's main-loop command: while this controller's local Raft state is `observer`,
/// admit it into the quorum's voter set once that is safe.
pub fn quorum_manager_container_command() -> String {
// The sidecar is a separate container and inherits nothing from the kafka container's
// startup, so it derives the realm itself. Harmless when krb5.conf is absent: only the
// Kerberos case has a `${env:KERBEROS_REALM}` placeholder for `config-utils` to resolve.
let set_realm_env = format!(
"KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH} 2>/dev/null) && export KERBEROS_REALM || true"
);
format!(
r#"
set -uo pipefail
Expand All @@ -238,9 +259,12 @@ pub fn quorum_manager_container_command() -> String {
[ -n "$POD_INDEX" ] || exit 0
{export_replica_id}
{extract_bootstrap_servers}
{set_realm_env}

if cp {config_dir}/{controller_properties_file} /tmp/{controller_properties_file} \
&& config-utils template /tmp/{controller_properties_file} \
&& cp {admin_client_source} {admin_client_config} \
&& config-utils template {admin_client_config} \
&& cat /tmp/{controller_properties_file} {admin_client_config} > {add_controller_config}; then
QUORUM_CLI={binary}
ADMIN_CLIENT_CONFIG={admin_client_config}
Expand All @@ -266,8 +290,10 @@ pub fn quorum_manager_container_command() -> String {
derive_pod_index = DERIVE_POD_INDEX,
export_replica_id = EXPORT_REPLICA_ID,
extract_bootstrap_servers = extract_bootstrap_servers_command(),
set_realm_env = set_realm_env,
config_dir = STACKABLE_CONFIG_DIR,
controller_properties_file = ConfigFileName::ControllerProperties,
admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH,
admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH,
add_controller_config = ADD_CONTROLLER_PROPERTIES_PATH,
cli_timeout = CLI_CALL_TIMEOUT_SECONDS,
Expand Down Expand Up @@ -378,6 +404,65 @@ mod tests {
use indoc::indoc;

use super::*;
use crate::controller::build::security::tests::{kerberos, plaintext};

#[test]
fn controller_command_exports_the_kerberos_realm_when_enabled() {
let command = controller_kafka_container_command(&kerberos(), vec![]);
assert!(command.contains("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*'"));
}

#[test]
fn controller_command_does_not_export_a_realm_without_kerberos() {
let command = controller_kafka_container_command(&plaintext(), vec![]);
assert!(!command.contains("KERBEROS_REALM"));
}

#[test]
fn controller_command_always_templates_the_jaas_file() {
// `jaas.properties` is always present in the ConfigMap (empty when Kerberos is off),
// so the copy is unconditional, matching `broker_start_command`.
for security in [kerberos(), plaintext()] {
let command = controller_kafka_container_command(&security, vec![]);
assert!(command.contains("cp /stackable/config/jaas.properties /tmp/jaas.properties"));
assert!(command.contains("config-utils template /tmp/jaas.properties"));
}
}

#[test]
fn quorum_manager_templates_the_admin_client_config() {
let command = quorum_manager_container_command();
assert!(
command.contains(
"cp /stackable/config/admin-client.properties /tmp/admin-client.properties"
)
);
assert!(command.contains("config-utils template /tmp/admin-client.properties"));
// It must connect with the *rendered* copy, not the raw ConfigMap file, or the
// `${env:...}` placeholders in `sasl.jaas.config` reach the JAAS parser verbatim.
assert!(command.contains("ADMIN_CLIENT_CONFIG=/tmp/admin-client.properties"));
assert!(!command.contains("ADMIN_CLIENT_CONFIG=/stackable/config/admin-client.properties"));
}

#[test]
fn quorum_manager_exports_the_kerberos_realm() {
// The sidecar is a separate container: it inherits nothing from the kafka container's
// startup, so it must derive $KERBEROS_REALM itself for `config-utils template` to
// resolve the principal.
let command = quorum_manager_container_command();
assert!(command.contains("KERBEROS_REALM"));
}

#[test]
fn controller_command_templates_the_admin_client_config_for_pre_stop() {
let command = controller_kafka_container_command(&kerberos(), vec![]);
assert!(
command.contains(
"cp /stackable/config/admin-client.properties /tmp/admin-client.properties"
)
);
assert!(command.contains("config-utils template /tmp/admin-client.properties"));
}

#[test]
fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() {
Expand Down Expand Up @@ -446,11 +531,11 @@ mod tests {
// `listeners`) via the same REPLICA_ID derivation used by the `kafka` container.
assert!(command.contains("export REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET))"));
assert!(command.contains("config-utils template /tmp/controller.properties"));
// Merges it with the plain admin-client config (carries `security.protocol`/`ssl.*`),
// controller.properties first so the client TLS config in admin-client.properties
// wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`'s doc comment).
// Merges it with the *rendered* admin-client config (carries `security.protocol`,
// `ssl.*` and, under Kerberos, `sasl.jaas.config`), controller.properties first so
// the client config wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`).
assert!(command.contains(
"cat /tmp/controller.properties /stackable/config/admin-client.properties > /tmp/add-controller.properties"
"cat /tmp/controller.properties /tmp/admin-client.properties > /tmp/add-controller.properties"
));
// The merged file is what `add-controller` — and only `add-controller` — connects
// with; read-only `describe` calls keep using the plain admin-client config.
Expand Down Expand Up @@ -1119,7 +1204,7 @@ mod tests {
pod_descriptor(KafkaRole::Controller, 1, 6),
pod_descriptor(KafkaRole::Controller, 2, 7),
];
let command = controller_kafka_container_command(descriptors);
let command = controller_kafka_container_command(&plaintext(), descriptors);

assert!(command.contains(r#"if [ "$REPLICA_ID" = "5" ]; then"#));
assert!(command.contains("FORMAT_QUORUM_FLAG=--standalone"));
Expand Down
Loading
Loading