Security

SASL and TLS are librdkafka properties set through config; the ssl and ssl-vendored features link the OpenSSL they need.

security.protocol picks the listener: PLAINTEXT, SSL, SASL_PLAINTEXT or SASL_SSL. sasl.mechanism, sasl.username and sasl.password set the SASL login, and the ssl.* properties the certificates. The bootstrap servers are the addresses of the listener the protocol speaks to.

On the default features, librdkafka carries one SASL mechanism, PLAIN. The ssl feature links the system OpenSSL, and ssl-vendored builds OpenSSL from source; either one adds TLS and the SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER mechanisms. Without it, a TLS property fails startup with "OpenSSL not available at build time", and a SCRAM or OAUTHBEARER mechanism fails the connect. The zstd feature adds that compression codec; the other codecs are built in.

A misspelled security.protocol fails startup, in a TestApp test as on a cluster. The generated AsyncAPI document reports the bootstrap addresses and never the credentials.

Example

The dependency line:

ruststream-rdkafka = { version = "0.7", features = ["ssl"] }

The service authenticates with SCRAM over TLS and verifies the cluster against its own certificate authority. The password comes from the environment.

Macros
Manual

examples/kafka_broker_security.rs
/// The cluster's `SASL_SSL` listener, with the service's own credentials.
fn broker(password: &str, ca: &str) -> KafkaBroker {
    KafkaBroker::new(["kafka-1.internal:9093", "kafka-2.internal:9093"])
        .default_group("orders-svc")
        .config("security.protocol", "SASL_SSL")
        .config("sasl.mechanism", "SCRAM-SHA-512")
        .config("sasl.username", "orders-svc")
        .config("sasl.password", password)
        .config("ssl.ca.location", ca)
}

/// The service, with the password and the certificate authority the deployment hands it.
fn service(password: &str, ca: &str) -> impl App + use<> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker(password, ca), |b| {
        b.include(place);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        &env::var("KAFKA_PASSWORD").unwrap_or_default(),
        "/etc/kafka/ca.pem",
    )
}

examples/manual/kafka_broker_security.rs
/// The cluster's `SASL_SSL` listener, with the service's own credentials.
fn broker(password: &str, ca: &str) -> KafkaBroker {
    KafkaBroker::new(["kafka-1.internal:9093", "kafka-2.internal:9093"])
        .default_group("orders-svc")
        .config("security.protocol", "SASL_SSL")
        .config("sasl.mechanism", "SCRAM-SHA-512")
        .config("sasl.username", "orders-svc")
        .config("sasl.password", password)
        .config("ssl.ca.location", ca)
}

/// The service, with the password and the certificate authority the deployment hands it.
fn service(password: &str, ca: &str) -> impl App + use<> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker(password, ca), |b| {
        b.include(subscriber("orders", Place).build());
    })
}

fn app() -> impl App {
    service(
        &env::var("KAFKA_PASSWORD").unwrap_or_default(),
        "/etc/kafka/ca.pem",
    )
}

Example​

Example