Connecting to a production cluster

A production cluster asks for authentication and TLS, and a service sets its producer's durability and its own timeouts.

SASL over TLS

src/main.rs

/// Where the cluster is and who the service is, as the deployment provides them.
struct Cluster {
    servers: String,
    username: String,
    password: String,
}

impl Cluster {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        Self {
            servers: read("KAFKA_SERVERS"),
            username: read("KAFKA_USERNAME"),
            password: read("KAFKA_PASSWORD"),
        }
    }
}

fn broker(cluster: Cluster) -> KafkaBroker {
    KafkaBroker::new([cluster.servers])
        .default_group("orders-svc")
        .config("security.protocol", "SASL_SSL")
        .config("sasl.mechanism", "SCRAM-SHA-512")
        .config("sasl.username", cluster.username)
        .config("sasl.password", cluster.password)
        .producer_config("acks", "all")
        .producer_config("enable.idempotence", "true")
        .connect_timeout(Duration::from_secs(10))
        .flush_timeout(Duration::from_secs(20))
}

Cargo.toml

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

The deployment provides the address and the credentials in environment variables. Every librdkafka property goes through KafkaBroker::config, which applies it to the consumers and the producer. security.protocol selects SASL over TLS, and sasl.mechanism the SASL mechanism. TLS needs librdkafka built with OpenSSL: the ssl feature links the system OpenSSL, and ssl-vendored builds one from source. SCRAM and OAUTHBEARER need the same feature, because librdkafka builds them on OpenSSL. SASL PLAIN over SASL_PLAINTEXT needs no feature. Kerberos (GSSAPI) is the gssapi feature of rust-rdkafka: add rdkafka to your dependencies with that feature. A cluster with TLS and no SASL takes security.protocol set to SSL, and the ssl.* properties name the certificates.

Durability and timeouts

producer_config sets a property on the producer only. acks=all makes the cluster confirm a record once every in-sync replica has it. enable.idempotence keeps a producer retry from writing a record twice. Every publish waits for the cluster's confirmation, so a successful publish means the cluster accepted the record.

The broker has timeouts of its own:

• connect_timeout bounds the check that the cluster is reachable at startup.
• flush_timeout bounds the flush of unsent records at shutdown.
• assignment_timeout bounds how long startup waits for the group to assign a subscription its partitions.

The service

Macros

src/main.rs

#[subscriber(KafkaTopic::new("orders").commit(Commit::Tracked), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { id: order.id }
}

fn service(broker: KafkaBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm)
            .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(broker(Cluster::from_env()))
}

Manual

src/main.rs

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn service(broker: KafkaBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber("orders", Confirm)
                .map_source(|_| KafkaTopic::new("orders").commit(Commit::Tracked))
                .reply()
                .build(),
        )
        .out_reply(Publish::default().queue_timeout(Duration::from_secs(5)));
    })
}

fn app() -> RustStream {
    service(broker(Cluster::from_env()))
}

fn main() -> ExitCode {
    run_main(app)
}

A publish waits for space in librdkafka's local queue when the queue is full. queue_timeout on the publish policy turns a longer wait into an error.

A subscription sets its own properties with KafkaTopic::config, and they win over the broker's. Two properties belong to the commit mode: enable.auto.commit and enable.auto.offset.store. Under Commit::Tracked and Commit::Transactional a subscription that sets either refuses to start, and the error names the property and the mode.

Testing the configuration

The tests start the production service with the production broker, given test credentials, under TestApp::start. The in-process Kafka validates the configuration the way librdkafka does, without dialling the cluster. A property name or value that librdkafka refuses fails the start with librdkafka's message:

Invalid value "SASL_TLS" for configuration property "security.protocol"

The tests need the same features as the service. librdkafka built without OpenSSL refuses SASL_SSL and SSL as values of security.protocol. Authentication itself happens only on a real cluster.