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
- Manual
#[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);
})
}
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());
})
}