Upgrading to 0.27

Every breaking change in krafka 0.27 with the old API, the new API and what to do: the Kafka handle, send/enqueue, recv, metrics.

This page maps the 0.26 API to 0.27, area by area. The complete list is the release's Breaking and Removed sections in CHANGELOG.md. Start with construction, then each client you use, then metrics and tests.

Building clients

Connection settings are on KafkaBuilder; each client is built from the handle and shares its connection pool.

use krafka::Kafka;
use krafka::auth::{AuthConfig, TlsConfig};
use std::time::Duration;

let kafka = Kafka::builder("broker:9093")
    .client_id("payments")
    .security(AuthConfig::sasl_scram_sha512("user", "secret").with_tls(TlsConfig::new()))
    .request_timeout(Duration::from_secs(30))
    .connect()
    .await?;

let producer = kafka.producer().build().await?;
let consumer = kafka.consumer("payments-group").build().await?;
let admin = kafka.admin();

producer.close().await?;
consumer.close().await?;
admin.close().await?;

protocol, network and metadata are private modules. BrokerInfo, PartitionInfo, TopicInfo, MetadataRecoveryStrategy, ProxyConfig, TimestampType and Compression are re-exported at the crate root.

Cargo features

The features are ring (default), rustls-aws-lc-rs, zstd, aws-msk, oauth-oidc, native-tls-roots, tls-encrypted-keys, unstable-protocol and test-broker.

Errors

New kinds: Closed, Wakeup, UnknownTopic, DeliveryTimeout { possibly_written } and OutOfOrderSequence. Branch on is_retriable(), is_fatal() and requires_abort() rather than on kinds where you can. See Error Handling.

Producer

delivery_timeout must be at least linger + request_timeout, or build() fails. A partitioner that returns a partition outside [0, partition_count) fails that send with Config.

Transactions

A commit after any failed send of the transaction refuses with TransactionAbortable, whether or not you awaited that send's handle; call abort(). abort() fails the transaction's buffered records instead of sending them.

use krafka::{Kafka, Record};

let kafka = Kafka::builder("localhost:9092").connect().await?;
let producer = kafka.producer().build_transactional("orders-tx").await?;

producer.begin()?;
producer.send(Record::new("orders", "o-1").key("c-7")).await?;
match producer.commit().await {
    Ok(()) => {}
    Err(e) if e.requires_abort() => producer.abort().await?,
    Err(e) => return Err(e),
}

Consumer

Behaviour to check in your code:

• Position. A partition's position is the next record poll/recv will hand out. It moves only when records are returned; committing never moves it.
• Seeking. seek and its variants fail with IllegalState for a partition that is not assigned. seek_to_beginning goes to the log start offset.
• Subscribing. subscribe() returns before the group is joined; poll() applies the assignment and calls your listener.
• Poll-interval expiry. The next poll() reports the partitions to on_partitions_lost once, rejoins and returns normally.

use krafka::Kafka;
use krafka::consumer::{OffsetAndMetadata, TopicPartition};

let kafka = Kafka::builder("localhost:9092").connect().await?;
let consumer = kafka.consumer("billing").build().await?;
consumer.subscribe(["invoices"]).await?;

while let Some(record) = consumer.recv().await? {
    let offsets = [(
        TopicPartition::new(record.topic.as_ref(), record.partition),
        OffsetAndMetadata::with_metadata(record.offset + 1, "billing-v2"),
    )];
    consumer.commit_offsets(offsets).await?;
}

Share consumer

Implicit mode accepts the previous delivery when the next poll()/recv() starts, and on close(). See Share Consumer.

Admin client

Every operation is one method taking krafka types and an *Options struct (Default, with a timeout). Multi-item operations return HashMap<Item, Result<T, KrafkaError>>.

list_consumer_groups and list_transactions return a result per broker. default_api_timeout and retry_backoff are set on the client kafka.admin() returns. See Admin Client.

Security

See Authentication.

Metrics and tracing

The fake broker

krafka::testing is outside semver.

See Testing.