Blog Home
Apache Kafka

Introducing the Confluent Rust Client for Apache Kafka

byJaisen Mathai, Staff Product Manager

We're excited to announce the general availability of Confluent's officially supported native Rust client for Apache Kafka®.

Our Rust client is open source under the Apache 2.0 license, and ready for production.

Two things make it different from the Rust Kafka clients you may have used before. It's native, and it is aligned with the Java client.

A native Rust client

The client is written in Rust and uses the Kafka protocol directly. It installs and builds with Cargo like any other Rust crate.

In practice, that means:

  • There's no system library such as librdkafka to install first: add the crate and build with Cargo. A C compiler is still needed, since the TLS (AWS-LC) and zstd dependencies compile their bundled C sources.
  • It's async from the start and runs on Tokio. send and poll are futures you .await.
  • Callbacks, like rebalance listeners, are async Rust traits.

Aligned with the Java client

Apache Kafka includes the Java client, and Java is the only client language that is part of the project. That means new features are always added to the Java client first, and other clients have historically had trouble keeping feature parity, ending up as second-class citizens in the Kafka ecosystem. Our new Rust client is designed to follow the Java client closely: the same configuration keys, the same method names, and the same behavior.

That has a few practical benefits:

  • Your existing knowledge carries over. If you've used the Java client, the Rust API will look familiar.
  • More consistency with documentation. The Apache Kafka documentation, Java examples and the answers you find online mostly apply as written.
  • Behavior is predictable. Defaults, delivery guarantees and rebalance semantics match Java's, so a Rust service and a Java service in the same system behave the same way.

What's included

  • Producer, with idempotence and transactions.
  • Consumer, using the KIP-848 consumer group protocol.
  • Admin client.
  • gzip, snappy, lz4 and zstd compression.

The consumer uses KIP-848 only, so it needs Apache Kafka 4.0 or newer brokers.

Getting started

For a step-by-step walkthrough, from creating a project to producing and consuming events, see Getting Started with Apache Kafka and Rust. For configuration, the API and more, see the Rust client documentation. The short version is below.

Add the crate and Tokio:

cargo add confluent-kafka tokio --features tokio/full

Here's a producer that sends a record and waits for the broker to acknowledge it:

use std::collections::HashMap;

use confluent_kafka::common::Error;
use confluent_kafka::common::serialization::StringSerializer;
use confluent_kafka::producer::{KafkaProducer, Producer, ProducerConfig, ProducerRecord};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let props = HashMap::from([(
        ProducerConfig::BOOTSTRAP_SERVERS_CONFIG.to_string(),
        "localhost:9092".to_string(),
    )]);
    let config = ProducerConfig::new(&props)?;
    let producer = KafkaProducer::new(
        config,
        Box::new(StringSerializer::new()),
        Box::new(StringSerializer::new()),
    )?;

    let record = ProducerRecord::with_key("orders".to_string(), Some("order-42".to_string()), Some("shipped".to_string()));

    // send() returns a future for the delivery. Await it when you need the offset.
    let delivery = producer.send(record).await?;
    let metadata = delivery.get().await?;
    println!("{}-{} @ {}", metadata.topic(), metadata.partition(), metadata.offset());

    producer.close().await
}

And a consumer that subscribes to the topic and reads from it:

use std::collections::HashMap;
use std::time::Duration;

use confluent_kafka::common::Error;
use confluent_kafka::common::serialization::StringDeserializer;
use confluent_kafka::consumer::{Consumer, ConsumerConfig, GroupProtocol, KafkaConsumer};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let props = HashMap::from([
        (ConsumerConfig::BOOTSTRAP_SERVERS_CONFIG.to_string(), "localhost:9092".to_string()),
        (ConsumerConfig::GROUP_ID_CONFIG.to_string(), "order-processor".to_string()),
        (ConsumerConfig::GROUP_PROTOCOL_CONFIG.to_string(), GroupProtocol::Consumer.to_string()),
    ]);
    let config = ConsumerConfig::new(&props)?;
    let mut consumer = KafkaConsumer::new(
        config,
        Box::new(StringDeserializer::new()),
        Box::new(StringDeserializer::new()),
    )?;

    consumer.subscribe_with_topics(vec!["orders".to_string()]).await?;

    loop {
        let records = match consumer.poll(Duration::from_millis(500)).await {
            Ok(records) => records,
            Err(e) if e.is_retriable_error() => continue,
            Err(e) => return Err(e),
        };
        for record in records {
            println!("{:?} = {:?}", record.key(), record.value());
        }
        consumer.commit_sync().await?;
    }
}

If you've written a Java producer or consumer, both of these should read like a direct translation. There are more examples on GitHub, including transactions, rebalance listeners and Avro.

Coming from another Rust client

If you're moving from a librdkafka-based crate such as rust-rdkafka, a few differences are worth knowing up front:

  • Config keys use Java names. For example, message.max.bytes in librdkafka is max.request.size in the Rust client.
  • isolation.level defaults to read_uncommitted, as in Java. librdkafka defaults to read_committed, so set it explicitly if you depend on that.
  • Idempotence is on by default, as in Java.

What's next

The client is open source and developed in the open: the code and the design proposals are public on the kafka-clients GitHub repository, and community feedback is appreciated.

Try it, and open an issue if something doesn't work the way you expect, or start a discussion if you have questions or ideas.