Сегменты, ключи сегмента и порядок

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

Пример - сервис заказов. Он проводит каждый заказ в журнал покупателя по четыре заказа одновременно и подтверждает каждую проводку квитанцией.

Как ключ выбирает сегмент

У каждой записи Kinesis есть ключ сегмента. Kinesis хеширует ключ алгоритмом MD5 и сохраняет запись в тот сегмент, в диапазон хеш-ключей которого попадает хеш. Kinesis сохраняет порядок записей внутри одного сегмента и не соблюдает порядка между сегментами. Поэтому записи с одним ключом остаются по порядку, а записи с разными ключами распределяются по сегментам.

Пропускная способность одного ключа - пропускная способность одного сегмента: сегмент принимает до 1 МБ или 1000 записей в секунду. Ключ, на который приходится большая часть трафика, делает свой сегмент пределом всего потока.

Порядок для каждого покупателя

Макросы

src/main.rs

#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "customer-orders")]
struct Order {
    id: u64,
    customer: String,
}

/// One line of a customer's journal.
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "customer-orders.journal")]
struct Entry {
    order: u64,
    customer: String,
}

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

/// Posts each order to the journal, under the customer as its partition key.
#[subscriber(
    KinesisStream::new("customer-orders").create_if_missing(2),
    workers(4, by_key)
)]
async fn post(
    order: &Order,
    Out(journal): Out<impl Publisher<Options = KinesisPublishOptions>, Journal>,
) -> HandlerOutcome {
    let entry = Entry {
        order: order.id,
        customer: order.customer.clone(),
    };
    if journal
        .message(&entry)
        .partition_key(order.customer.as_str())
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, order.customer, "order posted");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Order {
    id: u64,
    customer: String,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "customer-orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

/// One line of a customer's journal.
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Entry {
    order: u64,
    customer: String,
}

impl OutgoingDestination for Entry {
    type Form = FixedName;

    const DESTINATION: &'static str = "customer-orders.journal";
}

impl MessageHeaders for Entry {
    type Contract = NoHeaders;
}

impl MessageInfo for Entry {
    const NAME: &'static str = "Entry";
}

struct Journal;

impl OutSlot for Journal {
    const NAME: &'static str = "Journal";
    type Destination = Reads;
}

impl PublishedThrough<Journal> for Entry {}

/// Posts each order to the journal, under the customer as its partition key.
struct Post;

impl<Lines> Handle<Order, (), Outs<(Lines,)>> for Post
where
    Lines: OutEntry<Journal, Wire: Publisher<Options = KinesisPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Lines,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let entry = Entry {
            order: order.id,
            customer: order.customer.clone(),
        };
        if outs
            .get(Journal)
            .message(&entry)
            .partition_key(order.customer.as_str())
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(order.id, order.customer, "order posted");
        Ok(())
    }
}

Магазин публикует каждый заказ с покупателем в качестве ключа сегмента, поэтому заказы одного покупателя лежат в одном сегменте в том порядке, в каком их оформили. У потока customer-orders два сегмента, и подписка читает оба одновременно.

workers(4, by_key) запускает четыре партиции воркеров. Доставки с одним ключом выполняются друг за другом в одной партиции воркеров. Ключ доставки - её ключ сегмента, который она содержит в заголовке partition-key (PARTITION_KEY_HEADER). Поэтому post обрабатывает четыре заказа одновременно и никогда не обрабатывает два заказа одного покупателя в одно и то же время. В ручной форме ту же настройку задаёт workers_by_key. Простой workers(4) выполняет четыре доставки одновременно без учёта ключей, и тогда два заказа одного покупателя могут завершиться в любом порядке.

Как задать ключ записи

post записывает каждую проводку журнала с partition_key(..), шагом публикации. Проводка попадает в сегмент своего покупателя, поэтому читатели журнала тоже видят каждого покупателя по порядку. Этот шаг - настройка Kinesis, поэтому слот объявлен с ограничением Publisher<Options = KinesisPublishOptions>: только тогда публикация предлагает этот шаг.

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

/// Confirms each journal entry with a receipt.
#[subscriber(KinesisStream::new("customer-orders.journal").create_if_missing(1), reply)]
async fn confirm(entry: &Entry) -> Receipt {
    Receipt { order: entry.order }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Receipt {
    order: u64,
}

impl OutgoingDestination for Receipt {
    type Form = FixedName;

    const DESTINATION: &'static str = "receipts";
}

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

impl MessageInfo for Receipt {
    const NAME: &'static str = "Receipt";
}

/// Confirms each journal entry with a receipt.
struct Confirm;

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

Макросы

src/main.rs

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(post).out(Journal, Publish::default()).build();
        // One key for every receipt: one shard, so the receipts keep one order between them.
        b.include(confirm)
            .out_reply(Publish::default())
            .partition_key("receipts");
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

Вручную

src/main.rs

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                KinesisStream::new("customer-orders").create_if_missing(2),
                Post,
            )
            .workers_by_key(nonzero!(4usize))
            .build(),
        )
        .out(Journal, Publish::default())
        .build();
        // One key for every receipt: one shard, so the receipts keep one order between them.
        b.include(
            subscriber(
                KinesisStream::new("customer-orders.journal").create_if_missing(1),
                Confirm,
            )
            .reply()
            .build(),
        )
        .out_reply(Publish::default())
        .partition_key("receipts");
    })
}

fn main() -> ExitCode {
    run_main(|| {
        service(
            KinesisBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
        )
    })
}

У ответа нет вызова, к которому можно добавить шаг, поэтому точка монтирования задаёт его ключ в политике: .out_reply(Publish::default()).partition_key("receipts"). Тогда все квитанции попадают в один сегмент, и читатели видят квитанции в одном порядке ценой пропускной способности этого сегмента.

Публикация выбирает ключ в следующем порядке:

1. Шаг partition_key(..) публикации.
2. Заголовок partition-key, который содержит публикация: записанный вручную или пришедший с записью из другого брокера RustStream.
3. Ключ, который точка монтирования задала в политике.
4. Новый ключ для каждой записи, который распределяет записи по сегментам.

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

Разделение и слияние

Разделение закрывает сегмент и открывает два дочерних, которые делят его диапазон хеш-ключей. Слияние закрывает два соседних сегмента и открывает один дочерний. Закрытые родительские сегменты хранят свои записи, пока не истечёт срок хранения. Подписка заново перечисляет сегменты каждые десять секунд. Она запускает дочерний сегмент, только когда каждый его родитель прочитан до конца, поэтому записи одного ключа сохраняют порядок при изменении. Дочерний сегмент без контрольной точки открывается с самой старой записи, поэтому ничего, что записали в него до запуска читателя, не пропускается. Страница Сегменты, разделение и слияние показывает родительские и дочерние сегменты на схеме, а живые тесты крейта разделяют и сливают поток в tests/shards_kinesis.rs.

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

Тесты

Тесты в файле примера запускают приложение внутри процесса. У каждого потока Kinesis внутри процесса один сегмент, поэтому тест, который читает заказы из двух сегментов, работает на стенде:

cargo test
KINESIS_TEST_ENDPOINT=http://localhost:4566 cargo test -- --test-threads=1