KafkaTopic

KafkaTopic reads one topic in one consumer group, with its start offset, commit mode, assignment and lane key.

KafkaTopic::new takes the topic, and every other setting is optional. An unset setting keeps librdkafka's own default; the crate adds none of its own. group names the consumer group and overrides the broker's default_group. start is the start offset of a group that has committed nothing yet. commit is the commit mode, which decides what an ack does. assignment is the group's rebalance strategy, and lane_key is what keyed lanes are keyed by. config passes any other consumer property to librdkafka. The subscription's name is the topic: a test reads its deliveries under that name, and the AsyncAPI document reports that channel. A publish to the topic reaches every group that reads it, so a delayed retry copy goes back to the topic with no setting at the mount site. A topic name that Kafka refuses, or a name that starts with ^, stops the service at startup; a pattern is KafkaTopics::pattern.

Example

Every order of orders is confirmed by the group orders-workers, and the confirmation goes to the topic its type declares.

Macros
Manual

examples/kafka_topics.rs
/// Confirms every order of the topic `orders`, as a member of the group `orders-workers`.
#[subscriber(
    KafkaTopic::new("orders")
        .group("orders-workers")
        .start(StartOffset::Earliest)
        .commit(Commit::Tracked)
        .assignment(Assignment::CooperativeSticky)
        .config("fetch.min.bytes", "1024"),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: true,
    }
}

examples/manual/kafka_topics.rs
/// Confirms every order it reads.
struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation {
            id: order.id,
            accepted: true,
        }))
    }
}

fn app() -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // The topic `orders`, read as a member of the group `orders-workers`.
        let orders = KafkaTopic::new("orders")
            .group("orders-workers")
            .start(StartOffset::Earliest)
            .commit(Commit::Tracked)
            .assignment(Assignment::CooperativeSticky)
            .config("fetch.min.bytes", "1024");
        b.include(
            subscriber("orders", Confirm)
                .reply()
                .build()
                .map_source(|_| orders),
        );
    })
}

Example​

Example