Configuration for Rust Client for Kafka

Configure each client with a HashMap<String, String> 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.

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<dyn std::error::Error>> {
    let mut props: HashMap<String, String> = [
        ("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='<api_key>' password='<api_secret>';"),
    ]
    .into_iter()
    .map(|(k, v)| (k.to_string(), v.to_string()))
    .collect();

    // Producer
    let producer = KafkaProducer::<String, String>::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::<String, String>::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.
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.

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.