# Configuration for Rust Client for Kafka Configure each client with a `HashMap` of configuration properties, which you pass to `ProducerConfig::new`, `ConsumerConfig::new`, or `AdminClientConfig::new`. The client supports only SASL/PLAIN authentication. The client doesn’t support SASL/SCRAM, OAuth, or AWS IAM. ## Example configuration The following example shows a minimal configuration for a producer, consumer, and admin client connecting to Confluent Cloud with SASL/PLAIN authentication. ```rust use std::collections::HashMap; use confluent_kafka::admin::{AdminClient, AdminClientConfig}; use confluent_kafka::common::serialization::{StringDeserializer, StringSerializer}; use confluent_kafka::consumer::{ConsumerConfig, KafkaConsumer}; use confluent_kafka::producer::{KafkaProducer, ProducerConfig}; #[tokio::main] async fn main() -> Result<(), Box> { let mut props: HashMap = [ ("bootstrap.servers", "pkc-xxxxx.us-east-1.aws.confluent.cloud:9092"), ("security.protocol", "SASL_SSL"), ("sasl.mechanism", "PLAIN"), ("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required \ username='' password='';"), ] .into_iter() .map(|(k, v)| (k.to_string(), v.to_string())) .collect(); // Producer let producer = KafkaProducer::::new( ProducerConfig::new(&props)?, StringSerializer::new(), StringSerializer::new(), )?; // Admin client let admin = AdminClient::create(AdminClientConfig::new(&props)?)?; // Consumer: add the consumer group properties props.insert("group.id".to_string(), "my-consumer-group".to_string()); props.insert("group.protocol".to_string(), "consumer".to_string()); let consumer = KafkaConsumer::::new( ConsumerConfig::new(&props)?, StringDeserializer::new(), StringDeserializer::new(), )?; // Use the clients here. Ok(()) } ``` ## Common properties The following properties apply to the producer, consumer, and admin client. | Property | Default | Description | |------------------------------------------|----------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `bootstrap.servers` | Required. Must be defined. | A comma-separated list of host and port pairs that are the addresses of
the Kafka brokers. For example, `broker1:9092,broker2:9092`. The admin
client can use `bootstrap.controllers` instead. | | `client.dns.lookup` | `use_all_dns_ips` | How the client uses the DNS lookups of the bootstrap servers. With
`use_all_dns_ips`, the client tries each returned IP address in turn
until a connection succeeds. With
`resolve_canonical_bootstrap_servers_only`, the client resolves each
bootstrap address to its canonical name before the lookup. | | `metadata.recovery.strategy` | `rebootstrap` | What the client does when none of the brokers it knows is available.
With `rebootstrap`, the client repeats the bootstrap process using
`bootstrap.servers`. With `none`, the client doesn’t rebootstrap. | | `client.id` | Empty string | An optional identifier string to include in server-side request logging.
Use this to identify the application producing or consuming messages in
broker logs and metrics. | | `security.protocol` | `PLAINTEXT` | The protocol used to communicate with brokers. Valid values are
`PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, and `SASL_SSL`. | | `sasl.mechanism` | Required. Set to `PLAIN`. | The SASL mechanism used for authentication. The client supports only
`PLAIN`. If `GSSAPI` is set, the client can’t authenticate. | | `sasl.jaas.config` | Empty string | Java Authentication and Authorization Service (JAAS) login context
parameters for SASL connections, in the format used
by the Java Client. Required when using SASL/PLAIN authentication. | | `request.timeout.ms` | `30000` | The maximum amount of time the client waits for a response to a request,
in milliseconds. | | `metrics.recording.level` | `INFO` | The recording level for metrics sensors. Set to `INFO`, because all
built-in sensors use `INFO`-level recording. `DEBUG` and `TRACE`
are accepted but have no effect. The value is case-sensitive. | | `metrics.num.samples` | `2` | The number of samples maintained in the rolling window for windowed
statistics such as averages and rates. Must be at least `1`. | | `metrics.sample.window.ms` | `30000` | The window size for windowed statistics, in milliseconds. Must be at
least `0`. | | `retry.backoff.ms` | `100` | The initial time to wait before retrying a failed request, in
milliseconds. Backoff increases exponentially up to
`retry.backoff.max.ms`. | | `retry.backoff.max.ms` | `1000` | The maximum time to wait between retries, in milliseconds. | | `reconnect.backoff.ms` | `50` | The initial time to wait before reconnecting to a broker after the
connection is closed, in milliseconds. Backoff increases exponentially
up to `reconnect.backoff.max.ms`. | | `reconnect.backoff.max.ms` | `1000` | The maximum time to wait before reconnecting to a broker, in
milliseconds. | | `connections.max.idle.ms` | `300000` | The time after which idle connections are closed, in milliseconds.
The consumer default is `540000`. | | `metadata.max.age.ms` | `300000` | The maximum time before the client force-refreshes metadata, even if
no partition leadership changes have occurred, in milliseconds. | | `socket.connection.setup.timeout.ms` | `10000` | The time the client waits for a socket connection to be established,
in milliseconds. If the connection isn’t established within this time,
the client closes the connection and retries. | | `socket.connection.setup.timeout.max.ms` | `30000` | The maximum connection setup timeout, in milliseconds. The timeout
increases exponentially for each consecutive failure up to this ceiling. | | `send.buffer.bytes` | `131072` | The size of the TCP send buffer, in bytes. Set to `-1` to use the
operating system default. | | `receive.buffer.bytes` | `65536` | The size of the TCP receive buffer, in bytes. Set to `-1` to use the
operating system default. | ## TLS properties The following properties configure TLS when `security.protocol` is `SSL` or `SASL_SSL`, and apply to the producer, consumer, and admin client. The client reads keys and certificates only in PEM format. It doesn’t support JKS or PKCS12 stores, so convert them to PEM. It doesn’t support encrypted private keys, so the `ssl.key.password`, `ssl.keystore.password`, and `ssl.truststore.password` properties aren’t available. If you set neither `ssl.truststore.location` nor `ssl.truststore.certificates`, the client verifies the broker certificate against Mozilla’s root certificates, which the `webpki-roots` Cargo feature includes in the build. | Property | Default | Description | |-----------------------------------------|-------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `ssl.truststore.location` | None | The path to a PEM file with the certificates of the certificate
authorities that the client trusts. | | `ssl.truststore.certificates` | None | The certificates of the certificate authorities that the client trusts,
as a PEM string. Use it instead of `ssl.truststore.location`. | | `ssl.truststore.type` | `PEM` | The format of the trust store. The only supported value is `PEM`. | | `ssl.keystore.location` | None | The path to a PEM file with the client’s private key and certificate
chain, for mutual TLS authentication. | | `ssl.keystore.key` | None | The client’s private key, as a PEM string. Use it with
`ssl.keystore.certificate.chain` instead of
`ssl.keystore.location`. | | `ssl.keystore.certificate.chain` | None | The client’s certificate chain, as a PEM string. | | `ssl.keystore.type` | `PEM` | The format of the key store. The only supported value is `PEM`. The Java Client
default is `JKS`. | | `ssl.endpoint.identification.algorithm` | `https` | The algorithm the client uses to verify that the broker host name
matches the broker certificate. Set to an empty string to turn off
host name verification. | | `ssl.enabled.protocols` | `TLSv1.2,TLSv1.3` | The TLS protocol versions the client can use. | | `ssl.protocol` | `TLSv1.3` | The TLS protocol whose default protocol versions and cipher suites
apply when `ssl.enabled.protocols` or `ssl.cipher.suites` is empty. | | `ssl.cipher.suites` | None | A comma-separated list of the cipher suites the client can use, in
order of preference. Leave unset to use the default cipher suites. | ## Producer properties The following properties apply only to the producer (`ProducerConfig`). | Property | Default | Description | |-----------------------------------------|-------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `acks` | `all` | The number of acknowledgments the producer requires before considering a
request complete. Valid values are `0`, `1`, and `all`. | | `linger.ms` | `5` | The time the producer waits before sending a batch, in milliseconds.
Increasing this value allows more records to accumulate in the same
batch, which can improve throughput at the cost of latency. | | `delivery.timeout.ms` | `120000` | The maximum time the producer waits for a record to be acknowledged,
including retries, in milliseconds. | | `batch.size` | `16384` | The maximum number of bytes to include in a single batch per partition.
Larger values can improve throughput at the cost of memory. | | `buffer.memory` | `33554432` | The total memory the producer uses to buffer records waiting to be sent,
in bytes. `send()` blocks for up to `max.block.ms` when this limit
is reached. | | `max.block.ms` | `60000` | The maximum time `send()` blocks when the record buffer is full or
metadata is unavailable, in milliseconds. | | `max.request.size` | `1048576` | The maximum size of a single produce request, in bytes. | | `max.in.flight.requests.per.connection` | `5` | The maximum number of unacknowledged requests the producer sends to a
single broker before blocking. Set to `1` to ensure strict per-partition
ordering when `enable.idempotence` is `false`. | | `compression.type` | `none` | The compression type for all data the producer generates. Valid values
are `none`, `gzip`, `snappy`, `lz4`, and `zstd`. | | `retries` | `2147483647` | The number of times the producer retries sending a record after a
transient error. In practice, `delivery.timeout.ms` limits the time
spent retrying. | | `enable.idempotence` | `true` | When `true`, the producer ensures exactly-once delivery per partition.
Requires `acks=all` and `retries` greater than `0`. | | `partitioner.type` | `ConsistentRandomPartitioner` | The partitioner used for records that don’t specify a partition. Valid
values are `ConsistentRandomPartitioner`,
`Murmur2RandomPartitioner`, and `RoundRobinPartitioner`. The default
hashes keys with CRC-32, as the `consistent_random` partitioner of
librdkafka does, so a key can be assigned a different partition than
the Java Client assigns it. Set `Murmur2RandomPartitioner` to assign
the same partitions as the Java Client, which hashes keys with murmur2. | | `transactional.id` | None | The transactional ID to use for atomic writes across partitions and
topics. Required to use the transactional producer API. When set,
`enable.idempotence` is automatically enabled. | | `transaction.timeout.ms` | `60000` | The maximum lifetime of a transaction before the broker aborts it, in
milliseconds. | | `metadata.max.idle.ms` | `300000` | The time after which the producer drops cached metadata for topics that
haven’t produced any records, in milliseconds. | ## Consumer properties The following properties apply only to the consumer (`ConsumerConfig`). | Property | Default | Description | |-----------------------------|------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `group.id` | Must be defined. | The consumer group the consumer belongs to. Consumers with the same
`group.id` share the partitions of the topics they subscribe to.
Required to subscribe to topics and to commit or fetch committed
offsets. Without it, the consumer can read only the partitions that you
assign to it with `assign`. | | `group.protocol` | Required. Set to `consumer`. | The rebalance protocol to use. The `classic` group protocol returns
an error. See [Requirements](overview.md#rust-client-requirements). | | `auto.offset.reset` | `latest` | The offset-reset policy when there is no initial offset in Kafka or when
the current offset no longer exists. Valid values are `earliest`,
`latest`, and `none`. | | `enable.auto.commit` | `true` | When `true`, the consumer periodically commits offsets automatically.
Set to `false` to commit offsets manually with `commit_sync` or
`commit_async`. | | `max.poll.interval.ms` | `300000` | The maximum time between `poll()` invocations before the broker
considers the consumer dead and triggers a rebalance, in milliseconds.
Increase this value if record processing takes a long time. | | `max.poll.records` | `500` | The maximum number of records returned by a single call to `poll()`. | | `session.timeout.ms` | `45000` | Has no effect, because the broker controls the session timeout with
`group.consumer.session.timeout.ms`. | | `fetch.min.bytes` | `1` | The minimum amount of data the broker should return for a fetch request,
in bytes. Increasing this value reduces the number of fetch requests at
the cost of added latency. | | `fetch.max.wait.ms` | `500` | The maximum time the broker waits for `fetch.min.bytes` to be
available before responding, in milliseconds. | | `fetch.max.bytes` | `52428800` | The maximum number of bytes the server returns across all partitions for
a single fetch request. | | `max.partition.fetch.bytes` | `1048576` | The maximum number of bytes the server returns per partition in a fetch
response. | | `heartbeat.interval.ms` | `3000` | Has no effect, because the broker controls the heartbeat interval with
`group.consumer.heartbeat.interval.ms` and sends it to the consumer in
each heartbeat response. | | `auto.commit.interval.ms` | `5000` | The interval at which the consumer automatically commits offsets, in
milliseconds. Only applies when `enable.auto.commit` is `true`. | | `isolation.level` | `read_committed` | Controls which transactional records the consumer reads. `read_committed`
skips records from aborted transactions. Set to `read_uncommitted` to
receive all records regardless of transaction state. This default
differs from the Java Client, whose default is `read_uncommitted`. | | `client.rack` | Empty string | The rack of the consumer. The consumer sends it to the brokers so that
it can fetch from the closest replica, when the brokers have a replica
selector configured, and to the group coordinator for rack-aware
partition assignment. | | `allow.auto.create.topics` | `true` | When `true`, the consumer allows topics to be auto-created on the
broker when subscribing. | | `exclude.internal.topics` | `true` | When `true`, internal topics such as `__consumer_offsets` are
excluded from wildcard subscriptions. | | `check.crcs` | `true` | When `true`, the consumer automatically verifies the CRC32 checksum of
consumed records. Disable to improve throughput if your data pipeline
provides its own integrity checks. | | `group.instance.id` | None | A unique static identifier for this consumer instance. When set, the
consumer uses static group membership and doesn’t trigger a rebalance
on restart. | | `group.remote.assignor` | None | The name of the server-side assignor to use with the KIP-848 consumer
group protocol. Leave unset to use the broker default. | ## Admin client properties The following properties apply only to the admin client (`AdminClientConfig`). The admin client also accepts all [common properties](#rust-client-common-config). | Property | Default | Description | |--------------------------|--------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `bootstrap.controllers` | None | A comma-separated list of host and port pairs that are the addresses of
the Kafka controllers. Use it instead of `bootstrap.servers` to send
requests directly to the controllers. Set exactly one of the two
properties. | | `default.api.timeout.ms` | `60000` | The default timeout for admin client operations that don’t specify a
per-call timeout, in milliseconds. Must be greater than or equal to
`request.timeout.ms`. If set explicitly below `request.timeout.ms`,
client creation fails. | | `retries` | `2147483647` | The maximum number of times the admin client retries a transient request
failure before returning an error. |