Server

The server reports bare host:port coordinates and the schema registry, never credentials and never a protocol version.

A broker registered with with_broker_labeled is a server of the document, under its label. Its host is the bootstrap servers, joined with commas. A PLAINTEXT:// or SASL_SSL:// prefix and any user info an address carries stay out of it. The protocol is kafka. protocolVersion stays out: a Kafka client and the cluster agree on a version per request type, so no one number describes what clients speak.

A broker with a SchemaRegistry adds the server binding. schemaRegistryUrl is the registry's address without user info, and schemaRegistryVendor is confluent, the envelope this crate writes. The SASL password and the registry's basic-auth credentials stay out of the document.

Example

The broker reaches two brokers over SASL_SSL and a registry with basic auth. Its server in the document is kafka-1.internal:9093,kafka-2.internal:9093 with the registry https://registry.internal:8081.

Macros
Manual

examples/kafka_asyncapi.rs
/// The production cluster over `SASL_SSL`, with its schema registry.
fn broker(password: &str) -> KafkaBroker {
    let registry =
        SchemaRegistry::new("https://registry.internal:8081").basic_auth("orders-svc", password);
    KafkaBroker::new([
        "SASL_SSL://kafka-1.internal:9093",
        "SASL_SSL://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)
    .schema_registry(registry)
}

examples/manual/kafka_asyncapi.rs
/// The production cluster over `SASL_SSL`, with its schema registry.
fn broker(password: &str) -> KafkaBroker {
    let registry =
        SchemaRegistry::new("https://registry.internal:8081").basic_auth("orders-svc", password);
    KafkaBroker::new([
        "SASL_SSL://kafka-1.internal:9093",
        "SASL_SSL://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)
    .schema_registry(registry)
}

Example​

Example