Getting started

This guide installs ruststream-rdkafka, starts a local Kafka and runs a first service against it.

Installation

Cargo.toml

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

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

ruststream is the framework: macros gives the attributes, json the default codec, and logging prints the service's log to the terminal. ruststream-rdkafka is the Kafka broker. Its testing feature adds the in-process Kafka that tests run on.

The broker crate has these features:

Feature	What it gives
json (default), msgpack, cbor	the codec of the same name
schema-registry	the Confluent Schema Registry client and its wire format
avro	Avro over the Schema Registry
protobuf	Protobuf messages, with the Confluent envelope
asyncapi	the Kafka bindings in the generated AsyncAPI document
testing	the in-process Kafka behind TestApp
ssl	TLS through the system OpenSSL
ssl-vendored	TLS through an OpenSSL built from source
zstd	the zstd compression codec

System requirements

The crate builds librdkafka, the C client of Kafka, from source. The build needs a C compiler and make; librdkafka configures itself with its own script. On Debian and Ubuntu, build-essential provides both. The ssl feature also needs the OpenSSL headers (libssl-dev), and ssl-vendored needs perl to build OpenSSL. A CMake build or a system librdkafka is a feature of rust-rdkafka: add rdkafka to your own dependencies with the feature you need, and the build picks it up.

A local Kafka

One Kafka node in KRaft mode is enough to develop against:

compose.yaml

services:
  kafka:
    image: apache/kafka:4.0.0
    ports:
      - "9092:9092"

Start it and create the topic the service reads:

docker compose up -d
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --topic orders

The broker listens on localhost:9092.

The first service

Macros

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[subscriber("orders")]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[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(take_order);
        },
    )
}

Manual

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    amount_cents: u64,
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

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(subscriber("orders", TakeOrder).build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

take_order reads JSON orders from the orders topic. It acknowledges an order with an amount and drops one without. KafkaBroker::new takes the bootstrap servers and only records them. The connection opens when the service starts. Kafka reads a topic only inside a consumer group. default_group is the group of every subscription that names none. #[ruststream::app] turns app into main; in the manual form, run_main does the same.

Run the service:

cargo run -- run

Publish an order from another terminal with the console producer that ships in the image:

echo '{"id":1,"amount_cents":1250}' | docker compose exec -T kafka \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders

The service logs order taken with the order's id. Ctrl+C stops it.

cargo generate scaffolds the same kind of service with a router and a reply:

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

The first test

The example file also holds two tests. Each starts the same app with TestApp::start, publishes an order and checks how take_order settled it. KafkaBroker connects to an in-process Kafka there, so the tests need no cluster:

cargo test

Testing a Kafka service walks through these tests and what the in-process Kafka models.

Coming from rust-rdkafka

A service on rust-rdkafka writes its own consumer loop and producer. Here is where each part goes in RustStream:

rust-rdkafka	RustStream
ClientConfig::new().set("group.id", ..)	default_group, or KafkaTopic::group per subscription
ClientConfig::set for any other property	KafkaBroker::config, producer_config, KafkaTopic::config
StreamConsumer::subscribe and a recv loop	a #[subscriber] handler
deserializing message.payload()	the handler's parameter type, decoded by the codec
store_offset or commit_message after the work	Commit::Tracked and the handler's outcome (Commits and delivery guarantees)
FutureProducer::send	a reply or an Out parameter, sent through a publish policy
seek and assign	start_at at the mount site, or the seek handle in the handler (Replay and seeking)
send_offsets_to_transaction	EosPublish (Transactions and exactly-once)