Shutdown

Shutdown flushes in-flight publishes within flush_timeout, and a publisher used afterwards reports KafkaError::Closed.

librdkafka sends records in the background, so a publish the cluster has not yet confirmed may still sit in the producer's queue. At shutdown the runtime closes the connection first and then flushes that queue. The flush waits up to flush_timeout, 30 seconds by default. Records still queued after it fail the shutdown with KafkaError::Publish. A service that never published has no producer and nothing to flush.

A publisher taken before the shutdown keeps pointing at the closed connection. A publish through it, or a commit or an abort of its transaction, returns KafkaError::Closed with the topic or the transactional id, and never succeeds against the dead connection. A tool that drives the broker by hand receives a ClosedKafkaBroker from the shutdown; its unflushed_records is the number of records librdkafka still held.

Example

The flush gets ten seconds, and the shipments the handler replied with reach the cluster before the service exits.

Macros
Manual

examples/kafka_broker_shutdown.rs
/// The cluster at `servers`; shutdown waits up to ten seconds for in-flight records.
fn broker(servers: &str) -> KafkaBroker {
    KafkaBroker::new([servers])
        .default_group("orders-svc")
        .flush_timeout(Duration::from_secs(10))
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker("localhost:9092"))
}

examples/manual/kafka_broker_shutdown.rs
/// The cluster at `servers`; shutdown waits up to ten seconds for in-flight records.
fn broker(servers: &str) -> KafkaBroker {
    KafkaBroker::new([servers])
        .default_group("orders-svc")
        .flush_timeout(Duration::from_secs(10))
}

fn app() -> impl App<State = ()> {
    service(broker("localhost:9092"))
}

Example​

Example