Config

config and producer_config pass any librdkafka property to every client of the broker or to its producer only.

A property set with config reaches every consumer and the producer the broker creates. A property that applies to one side only is ignored by the other, as librdkafka ignores it. producer_config applies on top of config, to the producer alone: acks, compression.type, message.max.bytes, linger.ms. A descriptor's own config adds consumer properties for one subscription. The names and values are librdkafka's, listed in its CONFIGURATION.md.

librdkafka checks every property when the broker connects. An unknown name or a value it refuses fails startup with KafkaError::Connect, so a typo never reaches the first publish. The in-process cluster a TestApp test runs on reads the same properties. It refuses what librdkafka refuses, and a record over the producer's message.max.bytes is refused there as on a cluster.

Example

The broker names its clients orders-svc and compresses what it publishes. A shipment over 64 KiB is refused, and the order that produced it is settled as a retry.

Macros

examples/kafka_broker_config.rs

#[ruststream::app]
fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        // Every client: the consumers and the producer.
        .config("client.id", "orders-svc")
        .config("socket.keepalive.enable", "true")
        // The producer only.
        .producer_config("compression.type", "lz4")
        .producer_config("message.max.bytes", "65536");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(ship);
    })
}

Manual

examples/manual/kafka_broker_config.rs

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("orders-svc")
        // Every client: the consumers and the producer.
        .config("client.id", "orders-svc")
        .config("socket.keepalive.enable", "true")
        // The producer only.
        .producer_config("compression.type", "lz4")
        .producer_config("message.max.bytes", "65536");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Ship).reply().build());
    })
}