Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Темы, группы консьюмеров и порядок

Это руководство читает тему в группах консьюмеров, сохраняет порядок каждого ключа на параллельных партициях воркеров и размещает записи по партициям: по ключу, по номеру или по очереди.

Пример - сервис банковского реестра. Он проводит переводы по счетам, проверяет их на мошенничество, ведёт указатель своих записей и открывает дела по оспоренным переводам.

Порядок по счёту​

Kafka сохраняет порядок записей внутри одной партиции. Записи с одинаковым ключом попадают в одну партицию, поэтому порядок каждого ключа сохраняется. post читает переводы с ключом по счёту и обрабатывает до восьми одновременно, не нарушая этого порядка.

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) запускает восемь партиций воркеров. Доставки с одинаковым ключом партиционирования выполняются друг за другом на одной партиции воркеров. LaneKey::RecordKey выбирает партицию воркеров по ключу записи, поэтому два перевода одного счёта никогда не выполняются одновременно, даже внутри одной партиции. Значение по умолчанию, LaneKey::Partition, выбирает её по исходной партиции. Так обрабатываются и записи без ключа. В ручной форме та же настройка - workers_by_key.

Ключ записи приходит в обработчик заголовком kafka-partition-key (PARTITION_KEY_HEADER). AccountKey - контракт заголовков с одним полем под этим именем. Headers<AccountKey> читает ключ доставки, а with_headers(&key) записывает его ключом проводки. Поэтому и в bank.entries проводка попадает в партицию своего счёта, и читатели реестра видят каждый счёт по порядку. Перевод без ключа не проходит контракт заголовков, и его завершает политика декодирования, которая по умолчанию его отбрасывает.

Группы консьюмеров​

Группа консьюмеров читает каждую запись один раз, какому бы участнику ни досталась партиция. Каждая группа читает свою копию. screen читает те же переводы в собственной группе, поэтому видит каждый перевод, который видит post.

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 задаёт группу. Подписка, которая не называет группу, входит в default_group брокера. Так делает post. Kafka читает тему только в составе группы, поэтому сервис, у которого нет ни того, ни другого, не запускается.

StartOffset говорит, откуда начинает группа, у которой нет зафиксированного смещения для партиции. Так бывает при первом запуске и после того, как срок хранения удалил её смещение. Earliest начинает с самой старой записи, Latest - с конца журнала, а Committed, значение по умолчанию, оставляет выбор свойству librdkafka auto.offset.reset, которое равно latest.

Assignment выбирает, как группа делит партиции между участниками. CooperativeSticky при входе или выходе участника переносит только партиции, которые меняют владельца, а остальные продолжают работать.

Запуск ждёт, пока группа назначит этому участнику его партиции, но не дольше assignment_timeout (по умолчанию 60 секунд). После этого каждая запись, опубликованная в тему, доходит до подписки. Дайте каждой подписке сервиса свою группу. Две подписки на разные темы в одной группе вызывают перебалансировку друг у друга: когда входит вторая, первая на миг теряет свои партиции. Первый участник ничего не читает, пока группа заново не назначит партиции, а затем продолжает с того места, где остановился.

Несколько тем и шаблон​

KafkaTopics читает несколько тем через один консьюмер и одну группу. KafkaTopics::pattern читает все темы, имя которых подходит под регулярное выражение. Выражение должно начинаться с ^. archive читает все темы банка и запоминает, где лежит каждая запись.

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,
    }
}

Обработчик берёт координаты своей доставки из контекста: Topic, Partition и Offset. У всех подходящих тем один тип полезной нагрузки. archive декодирует любой JSON. Тема, созданная после запуска, входит в шаблон при следующем обновлении метаданных librdkafka.

KafkaPartitions читает названные партиции и не входит ни в какую группу. Перебалансировки у неё не бывает. Это подходит для одного процесса на партицию или для инструмента, который изучает одну партицию.

Размещение записей по партициям​

open_case превращает оспоренный перевод в дело в теме bank.cases. У этой темы три партиции: 0 и 1 поровну делят обычные дела, а партиция 2 хранит срочные.

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) закрепляет одну запись за партицией. Это шаг KafkaPublishSteps, поэтому слот ограничен Publisher<Options = KafkaOptions>, чтобы этот шаг был доступен. Закреплённая партиция важнее ключа записи, а ключ записи важнее партиционера librdkafka. Публикация в партицию, которой у темы нет, завершается ошибкой.

Приложение​

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) кладёт каждое дело, которое не называет партицию, в следующую из партиций 0 и 1. Запись без ключа librdkafka кладёт в случайную партицию, и одни консьюмеры оказываются загружены сильнее других. Цикл распределяет записи поровну. Запись с ключом или с закреплённой партицией сохраняет своё размещение. Число партиций в цикле фиксировано: оно равно числу партиций, которые цикл должен использовать, и не больше, чем есть у темы.

У подписки на набор тем нет одной темы, куда отправить копию для повтора, поэтому приложение называет её через ToSourceTopic. Это объясняет руководство Повторы и темы dead-letter.

Тесты в файле примера запускают приложение внутри процесса. Kafka внутри процесса даёт каждой теме одну партицию, поэтому тест размещения по нескольким партициям работает с кластером:

KAFKA_TEST_URL=localhost:9092 cargo test