Crate ruststream_rdkafka 

Apache Kafka broker for the RustStream messaging framework, backed by rdkafka and librdkafka.

A subscription is one consumer in one consumer group, and a publish is one record on the topic the outgoing message names. Kafka settles by committed position, so what ack and nack do is decided by the subscription’s Commit mode. A setting with no typed option here is a librdkafka property, set through the config passthroughs.

A service

A handler is an async fn over the decoded payload, mounted on a broker inside the application object, and the attribute writes the main:

use ruststream_rdkafka::prelude::*;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    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(handle);
        },
    )
}

cargo run -- run starts it. KafkaBroker::new only records the configuration; the runtime connects the broker at startup and shuts it down when the service stops.

Where things are

The guides take a Kafka service from installation to production, in this order: getting started, topics, consumer groups and ordering, commits and delivery guarantees, retries and dead-letter topics, replay and seeking, transactions and exactly-once, Schema Registry, Protobuf, testing and connecting to a production cluster. The architecture pages show how the crate sits between librdkafka and the runtime, and the benchmarks what it costs.

Each topic below links to its component page on the site, which shows it in a service; the item’s own documentation keeps the detail.

Broker

• Kafka broker: KafkaBroker, connected into ConnectedKafkaBroker and closed into ClosedKafkaBroker.
• Config: KafkaBroker::config and KafkaBroker::producer_config pass any librdkafka property.
• Security: SASL and TLS through config, with the ssl, ssl-vendored and zstd features.
• Startup: KafkaBroker::connect_timeout and KafkaBroker::assignment_timeout.
• Shutdown: KafkaBroker::flush_timeout and ClosedKafkaBroker::unflushed_records.
• Prelude: prelude, the one glob a Kafka service imports.
• Errors: KafkaError.

Subscriptions

• Subscriptions: a bare topic name, read in the broker’s default_group.
• KafkaTopic: KafkaTopic, one topic in one consumer group.
• KafkaTopics: KafkaTopics, a list of topics through one consumer.
• Patterns: KafkaTopics::pattern, every topic a regex matches.
• KafkaPartitions: KafkaPartitions, named partitions with no group.
• StartOffset: StartOffset, where a group with no committed offset starts.
• Assignment: Assignment, the group’s rebalance strategy.
• LaneKey: LaneKey, what a workers(n, by_key) lane is keyed by.
• Passthrough: KafkaTopic::config and its twins pass consumer properties.
• Batches: a slice handler over KafkaSubscriber batches.

Settlement

• Commit: Commit, what an ack or a nack does on Kafka.
• Auto: Commit::Auto, librdkafka’s timer commit.
• Tracked: Commit::Tracked, precise at-least-once.
• Retries: max_attempts and retry_after copies.
• Dead letters: dead_letter, the topic a spent delivery goes to.
• ToSourceTopic: ToSourceTopic, a retry copy back to its own topic.

Context

• Kafka context: KafkaContext and the keys in context::keys: topic, partition, offset, timestamp and key of a delivery.
• KafkaBatchContext: KafkaBatchContext, a batch handler’s context.

Positions

• KafkaPosition: KafkaPosition, a place in the log.
• start_at: opening a subscription at a position on every startup.
• Seek: SeekHandle and KafkaSeeker, moving a running subscription.

Publishing

• Publish: KafkaPublish, the plain publish policy, and its live KafkaPublisher.
• Keys: PARTITION_KEY_HEADER becomes the record’s native key.
• partition: KafkaOptions and KafkaPublishSteps::partition pin one record.
• RoundRobin: RoundRobin spreads replies over partitions.
• Durability: the producer’s acks and idempotence through KafkaBroker::producer_config.

Transactions

• Transactions: KafkaTransactionalPublish and its live KafkaTransactionalPublisher.
• per_partition: KafkaTransactionalPublish::per_partition, KafkaPartitionedPublish and PartitionLanes.
• Exactly once: KafkaEosPublish with Commit::Transactional.
• EosReplies: EosReplies joins replies to the open window.
• EosPipeline: EosPipeline and SourceOffset publish into it by hand.

Formats and tooling

• Schema Registry: schema_registry, the Confluent client and wire format (feature schema-registry).
• Avro: avro, the Avro codec (feature avro).
• Protobuf: protobuf, prost messages under the Confluent envelope (feature protobuf).
• AsyncAPI: the Kafka bindings of the generated document (feature asyncapi).
• Testing: the in-process Kafka cluster TestApp connects the broker to (feature testing), and Live runs against a running cluster.

Cargo features

json (default), msgpack and cbor forward the core’s codecs; the default codec is what a mount site with no .codec(..) step encodes with. schema-registry adds the Confluent client, the wire-format envelope and the subscriber-side prefetch; avro and protobuf add those formats on top of it. asyncapi contributes the Kafka bindings to the generated document, testing adds the in-process mode TestApp::start connects the broker through, and ssl, ssl-vendored and zstd map onto rdkafka’s own backends.

Re-exports

pub use protobuf::KafkaFramedPublish;

pub use protobuf::KafkaFramedPublisher;

pub use protobuf::ProtobufFrame;

pub use schema_registry::HttpRegistryClient;

pub use schema_registry::MemorySchemaCache;

pub use schema_registry::MissingSubject;

pub use schema_registry::RegisteredSchema;

pub use schema_registry::RegistryClient;

pub use schema_registry::SchemaCache;

pub use schema_registry::SchemaCachePolicy;

pub use schema_registry::SchemaFrame;

pub use schema_registry::SchemaFramed;

pub use schema_registry::SchemaPrefetch;

pub use schema_registry::SchemaRegistry;

pub use schema_registry::SchemaType;

pub use schema_registry::SubjectStrategy;

Modules

avro

Avro behind the codec.

context

Per-delivery and per-batch context fields exposed to handlers.

prelude

The imports a service on Kafka writes every time, in one glob.

protobuf

Protobuf as a self-serializing type under the Confluent envelope.

schema_registry

The Confluent Schema Registry: the client, the wire-format envelope, and the two ways a service speaks it.

Structs

ClosedKafkaBroker

The terminal witness returned by ConnectedBroker::shutdown.

ConnectedKafkaBroker

The connected form of KafkaBroker: the typed witness that Broker::connect succeeded.

EosPipeline

An exactly-once pipeline over one transactional producer, the live form of KafkaEosPublish.

EosReplies

The PublishTransform relaying EOS_SOURCE_HEADER from the originating delivery onto the reply, so the pipeline’s Publisher impl can pair the reply with its consumed offset.

KafkaBroker

An Apache Kafka broker backed by rdkafka / librdkafka.

KafkaEosPublish

The publish policy of EosPipeline: the pipeline id (the producer’s transactional id) plus the commit interval, declared anywhere and paired with the connected broker at startup.

KafkaMessage

One Kafka delivery: an owned snapshot of the record plus its settlement handle.

KafkaOptions

What one Kafka record carries beyond its topic, payload and headers.

KafkaPartitionedPublish

The publish policy of TransactionalPartitions, reached from KafkaTransactionalPublish::per_partition.

KafkaPartitions

A subscription that takes exactly the partitions it names, without joining a consumer group and without rebalancing.

KafkaPublish

The publish policy of KafkaPublisher: pure declaration, no connection, no publish surface.

KafkaPublisher

A live producer handle on the broker’s shared producer.

KafkaSeeker

Repositions a live KafkaSubscriber, minted by Seekable::seeker.

KafkaSubscriber

A consumer-group member on one topic, yielding KafkaMessage deliveries.

KafkaTopic

A subscription to one Kafka topic through one consumer group.

KafkaTopics

A subscription to several topics, or to every topic matching a regex, through one consumer and one consumer group.

KafkaTransactionalPublish

The publish policy of KafkaTransactionalPublisher: the transactional mode as its own type, reached from KafkaPublish::transactional_id.

KafkaTransactionalPublisher

A live publisher that produces inside Kafka transactions.

RoundRobin

A PublishTransform distributing records round-robin across the first count partitions.

SourceOffset

The source coordinates of one delivery, as EosPipeline::publish needs them.

ToSourceTopic

A PublishTransform sending each retry copy back to the topic its delivery arrived on.

TransactionalPartitions

Transactional publishers, one per source partition, materialized on first use.

Enums

Assignment

The partition assignment strategy for the consumer group (librdkafka’s partition.assignment.strategy).

Commit

How processed deliveries are committed back to the consumer group.

KafkaError

Errors returned by KafkaBroker and the types it hands out.

KafkaPosition

Where a subscription should resume reading.

LaneKey

What drives keyed worker lanes (workers(n, by_key)) for this subscription.

StartOffset

Where a consumer group starts reading when it has no valid committed offset.

Constants

EOS_SOURCE_HEADER

Header carrying a transactional delivery’s source coordinates through the reply path.

PARTITION_KEY_HEADER

Header carrying a message’s partition key, mapped onto Kafka’s native record key.

Traits

KafkaPublishSteps

The per-message steps this crate adds to the publish builder.

PartitionLanes

The capability of handing out one transactional publisher per source partition.