Skip to main content

Topics, consumer groups and ordering

This guide reads a topic in consumer groups, keeps the order of each key with parallel lanes, and places records on partitions by key, by number or in turn.

The example is a bank's ledger service. It posts transfers to account ledgers, screens them for fraud, keeps an index of its records and opens cases for disputed transfers.

Order per account​

Kafka keeps the order of records inside one partition. Records with the same key land on the same partition, so the order of each key holds. post reads transfers keyed by account and handles up to eight at once without breaking that order.

src/main.rs
/// The account a record belongs to, carried as the Kafka record key.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct AccountKey {
    #[serde(rename = "kafka-partition-key")]
    account: String,
}

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

/// One line of an account's ledger, keyed by the account like the transfer it came from.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "bank.entries", headers = AccountKey)]
struct Entry {
    transfer_id: u64,
    amount_cents: i64,
}

#[derive(OutSlot)]
#[publishes(Entry)]
struct Entries;

#[subscriber(
    KafkaTopic::new("transfers").lane_key(LaneKey::RecordKey),
    workers(8, by_key)
)]
async fn post(
    transfer: &Transfer,
    Headers(key): Headers<AccountKey>,
    Out(entries): Out<impl Publisher, Entries>,
) -> HandlerOutcome {
    let entry = Entry {
        transfer_id: transfer.id,
        amount_cents: transfer.amount_cents,
    };
    if entries
        .message(&entry)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    tracing::info!(transfer.id, account = %key.account, "transfer posted");
    HandlerOutcome::ack()
}

workers(8, by_key) runs eight lanes; deliveries with the same lane key run one after another on one lane. LaneKey::RecordKey keys the lanes by the record key, so two transfers of one account never run at the same time, even inside one partition. The default, LaneKey::Partition, keys them by the source partition, which also covers records without a key. In the manual form the same setting is workers_by_key.

The record key reaches the handler as the kafka-partition-key header (PARTITION_KEY_HEADER). AccountKey is a header contract with one field under that name. Headers<AccountKey> reads the key of the delivery, and with_headers(&key) writes it as the key of the ledger entry. So the entry goes to the partition of its account in bank.entries too, and readers of the ledger see each account in order. A transfer without a key fails the header contract and is settled by the decode policy, which drops it by default.

Consumer groups​

A consumer group reads each record once, whichever of its members gets the partition. Every group reads its own copy. screen reads the same transfers in a group of its own, so it sees every transfer that post sees.

src/main.rs
#[subscriber(
    KafkaTopic::new("transfers")
        .group("fraud-check")
        .start(StartOffset::Earliest)
        .assignment(Assignment::CooperativeSticky)
)]
async fn screen(transfer: &Transfer) -> HandlerOutcome {
    if transfer.amount_cents > 1_000_000 {
        tracing::warn!(transfer.id, "large transfer flagged");
    }
    HandlerOutcome::ack()
}

KafkaTopic::group names the group. A subscription that names none joins the broker's default_group; post does. Kafka reads a topic only inside a group, so a service with neither fails at startup.

StartOffset says where a group starts when it has no committed offset for a partition. That is the case on its first run and after retention deleted its offset. Earliest starts at the oldest record, Latest at the end of the log, and the default Committed leaves the choice to librdkafka's auto.offset.reset, which is latest.

Assignment picks how the group shares partitions among its members. CooperativeSticky moves only the partitions that change owner when a member joins or leaves, so the others keep running.

Startup waits until the group has assigned this member its partitions, up to assignment_timeout (60 seconds by default). After that, every record published to the topic reaches the subscription. Give each subscription of a service its own group. Two subscriptions of different topics in one group rebalance each other: when the second joins, the first loses its partitions for a moment. The first member reads nothing until the group has assigned the partitions again, and then continues where it stopped.

Several topics and a pattern​

KafkaTopics reads several topics through one consumer and one group. KafkaTopics::pattern reads every topic whose name matches a regular expression, which must start with ^. archive reads every topic of the bank and records where each record sits.

src/main.rs
/// Where a record of one of the bank's topics sits.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "archive.index")]
struct Archived {
    topic: String,
    partition: i32,
    offset: i64,
}

#[subscriber(KafkaTopics::pattern("^bank\\..*").group("archive"), reply)]
async fn archive(
    _record: &serde_json::Value,
    Ctx(topic): Ctx<Topic>,
    Ctx(partition): Ctx<Partition>,
    Ctx(offset): Ctx<Offset>,
) -> Archived {
    Archived {
        topic: topic.to_string(),
        partition,
        offset,
    }
}

The handler reads the coordinates of its delivery from the context: Topic, Partition and Offset. All matched topics share one payload type; archive decodes any JSON. A topic created after startup joins the pattern on librdkafka's next metadata refresh.

KafkaPartitions reads the partitions it names and joins no group. It never rebalances, which suits one process per partition or a tool that inspects one partition.

Placing records on partitions​

open_case turns a disputed transfer into a case on bank.cases. That topic has three partitions: 0 and 1 share the regular cases evenly, and partition 2 holds the urgent ones.

src/main.rs
/// The partition of `bank.cases` kept for urgent cases.
const URGENT: i32 = 2;

#[derive(Debug, PartialEq, Deserialize)]
struct Dispute {
    transfer_id: u64,
    urgent: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "bank.cases")]
struct Case {
    transfer_id: u64,
}

#[derive(OutSlot)]
#[publishes(Case)]
struct Cases;

#[subscriber(KafkaTopic::new("card-disputes").group("case-intake"))]
async fn open_case(
    dispute: &Dispute,
    Out(cases): Out<impl Publisher<Options = KafkaOptions>, Cases>,
) -> HandlerOutcome {
    let case = Case {
        transfer_id: dispute.transfer_id,
    };
    let opened = if dispute.urgent {
        cases.message(&case).partition(URGENT).publish().await
    } else {
        cases.message(&case).publish().await
    };
    if opened.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

partition(n) pins one record to a partition. It is a step of KafkaPublishSteps, so the slot is bounded Publisher<Options = KafkaOptions> to offer it. A pinned partition wins over the record key, and the record key wins over librdkafka's partitioner. A partition the topic does not have fails the publish.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(KafkaBroker::new(["localhost:9092"]).default_group("ledger"))
}

fn service(broker: KafkaBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        b.include(post).out(Entries, Publish::default()).build();
        b.include(screen);
        // A set of topics has no single topic to send a retry copy to, so the mount site names it.
        b.include(archive)
            .out_retry(Publish::default())
            .transform(ToSourceTopic);
        // Partitions 0 and 1 share the regular cases evenly; partition 2 stays for urgent ones.
        b.include(open_case)
            .out(Cases, Publish::default())
            .transform(RoundRobin::partitions(2))
            .build();
    })
}

RoundRobin::partitions(2) places each case that names no partition on the next of partitions 0 and 1. librdkafka places a record without a key on a random partition, which leaves some consumers busier than others; the cycle spreads the records evenly. A record with a key or a pinned partition keeps its placement. The count is fixed: it matches the partitions the cycle should use, never more than the topic has.

A subscription over a set of topics has no single topic to send a retry copy to, so the app names one with ToSourceTopic; Retries and dead-letter topics explains it.

The tests in the example file run the app in process. The in-process Kafka gives every topic one partition, so the test of placement on several partitions runs against a cluster:

KAFKA_TEST_URL=localhost:9092 cargo test