Crate autumn_plugin_kafka 

Kafka producer and consumers for Autumn apps.

Quick start

use autumn_plugin_kafka::{Consumer, HandlerError, KafkaPlugin, KafkaProducer, Message, Record};
use autumn_web::prelude::*;

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

#[post("/orders/{id}")]
async fn create(producer: KafkaProducer, Path(id): Path<u64>) -> AutumnResult<&'static str> {
    producer
        .send(Record::json("orders", &Order { id })?.with_key(id.to_string()))
        .await?;
    Ok("queued")
}

async fn on_order(msg: Message, _state: AppState) -> Result<(), HandlerError> {
    let order: Order = msg.json()?;
    tracing::info!(id = order.id, "order received");
    Ok(())
}

#[autumn_web::main]
async fn main() {
    autumn_web::app()
        .plugin(
            KafkaPlugin::new().consumer(
                Consumer::new("orders", ["orders"])
                    .group_id("billing")
                    .handler(on_order)
                    .dead_letter_topic("orders.dlq"),
            ),
        )
        .routes(routes![create])
        .run()
        .await;
}

Parts

Part	Function
KafkaPlugin	Installs all parts. Reads [kafka] at startup.
KafkaConfig	The [kafka] config section.
KafkaProducer	Sends records. It is a handler argument.
Consumer	A consumer with an async handler.
MemoryBroker	An in-memory broker for tests.
RdKafkaBackend	The default client, librdkafka.

The plugin also adds the kafka health indicator and kafka_* metrics to /actuator/prometheus.

Delivery

Consumers are at-least-once. The plugin commits an offset only after the handler completes, or after the dead-letter send completes. A handler can get the same message again. Make sure that this is safe.

Example

examples/orders.rs uses all parts. CI runs its self-check.

Structs

Consumer

A consumer: topics, a group, and a handler.

ConsumerSettings

Consumer settings. Maps to [kafka.consumer].

ConsumerSpec

The data that a backend needs to make one consumer.

Delivery

The position of a record after the broker accepted it.

HealthSettings

Health indicator settings. Maps to [kafka.health].

KafkaConfig

Settings for the plugin. Maps to the [kafka] section.

KafkaPlugin

The Kafka plugin.

KafkaProducer

Sends records to Kafka. Clones share one client.

KafkaRuntime

A handle to the running plugin. Clones share one runtime.

MemoryBroker

An in-memory broker. Use it in tests. It does not need Docker.

Message

A message received from Kafka.

ProducerSettings

Producer settings. Maps to [kafka.producer].

RdKafkaBackend

The default backend. It uses librdkafka.

Record

A record to send to Kafka.

Enums

ClientRole

The kind of Kafka client that a property map is for.

KafkaError

An error from the Kafka plugin.

Constants

DLQ_HEADER_CONSUMER

Dead-letter header: the consumer name.

DLQ_HEADER_ERROR

Dead-letter header: the last handler error. Maximum 1024 bytes.

DLQ_HEADER_OFFSET

Dead-letter header: the source offset, as decimal text.

DLQ_HEADER_PARTITION

Dead-letter header: the source partition, as decimal text.

DLQ_HEADER_TOPIC

Dead-letter header: the source topic.

PLUGIN_NAME

The plugin name.

Traits

Backend

Makes producers and consumers.

ConsumerBackend

Receives messages for one consumer group.

ProducerBackend

Sends records.

Type Aliases

BoxFuture

The future type of the backend traits. An owned dynamically typed Future for use in cases where you can’t statically type your result or need to add some indirection.

HandlerError

The error type that a handler can return.

Header

A header name and value.