ruststream-rdkafka 0.7.1

ruststream-rdkafka connects a RustStream service to Apache Kafka over rdkafka, the librdkafka binding. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• Consumer groups as descriptors: one topic, a set of topics or a regex, or partitions assigned by hand.
• Three commit modes: librdkafka auto-commit, precise per-message acknowledgement, and commits inside a producer transaction.
• Record keys and placement: the partition key becomes the Kafka key, and a publish can pin a record to a partition.
• Native batches cut from what librdkafka has already fetched.
• Retry caps and dead-letter topics declared where the handler is mounted.
• Transactions and exactly-once pipelines as publish policies.
• Repositioning: a handler moves its subscription to an offset or a timestamp.
• Confluent Schema Registry framing, with Avro and Protobuf, behind features.
• AsyncAPI with the specification's kafka binding, behind the asyncapi feature.
• Tests on the production app: the app main runs, with KafkaBroker connected to an in-process Kafka, or against a real cluster with TestApp::start_live.

Install

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-rdkafka = "0.7"
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
ruststream-rdkafka = { version = "0.7", features = ["testing"] }

The crate builds librdkafka from source, so a C toolchain is required. Optional features: testing, asyncapi, schema-registry, avro, protobuf, msgpack, cbor, ssl, ssl-vendored and zstd.

Write a service

use ruststream_rdkafka::prelude::*;
use serde::{Deserialize, Serialize};

// `Outgoing`, `PartialEq` and `Serialize` are here for the test below, which publishes an order
// and asserts on the decoded one.
#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
struct Order {
    id: u64,
}

#[derive(Debug, Deserialize, Outgoing, PartialEq, Serialize)]
struct Confirmation {
    id: u64,
}

#[subscriber(
    KafkaTopic::new("orders").commit(Commit::Tracked),
    reply("confirmations")
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("orders-svc"),
        |b| {
            b.include(confirm);
        },
    )
}

#[ruststream::app] generates main, so the binary understands run and asyncapi gen.

Scaffold a fresh project from the template:

cargo generate --git https://github.com/powersemmi/ruststream-rdkafka templates/kafka-topic --name my-service

Test it

TestApp runs the app main runs, with KafkaBroker connected to an in-process Kafka and no cluster.

use ruststream::testing::TestApp;

let tb = TestApp::start(app()).await?;

// `publish` returns once the handlers it woke have settled, so the assertions read finished state.
tb.broker::<KafkaBroker>()
    .message(&Order { id: 42 })
    .to("orders")
    .publish()
    .await?;

tb.broker::<KafkaBroker>()
    .subscriber("orders")
    .assert_called_once()
    .settled(HandlerOutcome::ack());

tb.broker::<KafkaBroker>()
    .published::<Confirmation>("confirmations")
    .assert_called_once()
    .with(&Confirmation { id: 42 });

Documentation

• Getting started - installation and a first service against a local Kafka.
• Guides - consumer groups and ordering, commits, retries, replay, exactly-once, Schema Registry, Protobuf, testing and a production cluster.
• Components - one page per part of the crate: the broker, subscriptions, settlement, publishing, transactions, the registry formats and testing.
• Architecture - how the crate sits between librdkafka and the runtime.
• Benchmarks - what the crate costs over the raw client.
• API reference - every item on docs.rs.
• The framework: https://powersemmi.github.io/ruststream/latest/

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

License

Licensed under the Apache-2.0 license.