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

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

Транзакции и exactly-once

Это руководство публикует несколько записей одной транзакцией Kafka, запускает транзакции при пуле воркеров и делает обработчик по схеме «прочитать, преобразовать, записать» exactly-once.

Транзакция Kafka делает набор записей видимым целиком. Читатель с уровнем изоляции read_committed, который в librdkafka стоит по умолчанию, видит их все после фиксации транзакции и не видит ни одной после прерывания. В RustStream транзакция - свойство политики публикации в точке монтирования, а обработчик называет только ту совместимость, которой пользуется.

Всё или ничего из одного обработчика​

ship превращает оплаченный заказ в отгрузочную строку на каждую позицию. Заказ с пустой строкой отбрасывается, и ни одна его строка не должна попасть на склад.

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    items: Vec<Item>,
}

#[derive(Debug, PartialEq, Deserialize)]
struct Item {
    sku: String,
    quantity: u32,
}

/// One shipment line per item of a paid order.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipment-lines")]
struct ShipmentLine {
    order_id: u64,
    sku: String,
    quantity: u32,
}

#[derive(OutSlot)]
#[publishes(ShipmentLine)]
struct Shipments;

#[subscriber(KafkaTopic::new("paid-orders").group("shipping").commit(Commit::Tracked))]
async fn ship(
    order: &Order,
    Out(shipments): Out<impl TransactionalPublisher, Shipments>,
) -> HandlerOutcome {
    if shipments.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for item in &order.items {
        if item.quantity == 0 {
            tracing::warn!(order.id, sku = %item.sku, "order with an empty line dropped");
            return match shipments.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        let line = ShipmentLine {
            order_id: order.id,
            sku: item.sku.clone(),
            quantity: item.quantity,
        };
        if shipments.message(&line).publish().await.is_err() {
            shipments.abort().await.ok();
            return HandlerOutcome::retry();
        }
    }
    if shipments.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, lines = order.items.len(), "order shipped");
    HandlerOutcome::ack()
}

Обработчик принимает свой слот как Out<impl TransactionalPublisher, Shipments>: он называет совместимость ядра TransactionalPublisher и ничего из Kafka. Записи, опубликованные между begin_transaction и commit, становятся видимыми вместе. abort отменяет их, поэтому первая строка заказа с пустой строкой так и не доходит до читателя. Вне открытой транзакции слот публикует как обычный. Второй begin_transaction при открытой транзакции сообщает KafkaError::TransactionBusy, а commit или abort, когда открытой транзакции нет, сообщает KafkaError::NoTransaction.

Транзакция охватывает опубликованные строки, но не прочитанный заказ. Commit::Tracked фиксирует смещение заказа после того, как обработчик его подтвердил. Если процесс упадёт между этими двумя шагами, заказ будет доставлен снова, и его строки уйдут во второй раз. Этот разрыв закрывает конвейер exactly-once, о котором речь ниже.

Транзакции при пуле воркеров​

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

src/main.rs
/// One invoice line per item of a billed order.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoice-items")]
struct InvoiceItem {
    order_id: u64,
    sku: String,
    quantity: u32,
}

#[subscriber(
    KafkaTopic::new("billable-orders").group("invoicing").commit(Commit::Tracked),
    workers(4, by_key)
)]
async fn invoice(
    order: &Order,
    Ctx(partition): Ctx<Partition>,
    Out(lanes): Out<impl PartitionLanes>,
) -> HandlerOutcome {
    let Ok(items) = lanes.for_partition(partition).await else {
        return HandlerOutcome::retry();
    };
    if items.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for item in &order.items {
        let line = InvoiceItem {
            order_id: order.id,
            sku: item.sku.clone(),
            quantity: item.quantity,
        };
        if items.message(&line).publish().await.is_err() {
            items.abort().await.ok();
            return HandlerOutcome::retry();
        }
    }
    if items.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, partition, "order invoiced");
    HandlerOutcome::ack()
}

workers(4, by_key) выбирает партицию воркеров по партиции записи, поэтому одна партиция обрабатывается по порядку на одной партиции воркеров. Слот - PartitionLanes: for_partition возвращает транзакционного издателя партиции доставки и создаёт его при первом обращении. Партиции воркеров, которые обрабатывают разные партиции, ведут свои транзакции одновременно. Транзакционные идентификаторы следуют за партициями темы, а не за числом воркеров, поэтому изменение workers(n) не меняет ни одного идентификатора.

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

Конвейер exactly-once​

enrich читает размещённый заказ и отвечает итогами, по которым планирует склад. Каждый ответ публикуется ровно один раз, даже если сервис упадёт и заказ будет прочитан снова.

src/main.rs
/// A placed order with the totals the warehouse plans with.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "enriched-orders")]
struct EnrichedOrder {
    id: u64,
    lines: usize,
    units: u32,
}

#[subscriber(
    KafkaTopic::new("placed-orders")
        .group("enrichment")
        .commit(Commit::Transactional("enrichment-1".into())),
    reply,
    workers(4, by_key)
)]
async fn enrich(order: &Order) -> EnrichedOrder {
    let units = order.items.iter().map(|item| item.quantity).sum();
    tracing::info!(order.id, units, "order enriched");
    EnrichedOrder {
        id: order.id,
        lines: order.items.len(),
        units,
    }
}

Обработчик только возвращает ответ, кода транзакций в нём нет. Commit::Transactional("enrichment-1") запрещает подписке фиксировать смещения самой. Вместо неё их фиксирует конвейер с тем же идентификатором, внутри транзакции, которая несёт ответы. Поэтому прочитанная позиция и опубликованные записи сдвигаются вместе: фиксируются либо оба, либо ни одно.

Точка монтирования​

Точка монтирования называет политику за каждым слотом и ответом.

src/main.rs
fn service(broker: KafkaBroker) -> impl App {
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(broker, |b| {
        // One transactional id per service instance.
        b.include(ship)
            .out(Shipments, Publish::default().transactional_id("shipping-1"))
            .build();
        // One transactional id per source partition: "invoicing-1-p0", "invoicing-1-p1", ...
        b.include(invoice)
            .out(
                DefaultSlot,
                Publish::default()
                    .transactional_id("invoicing-1")
                    .per_partition(),
            )
            .build();
        // The pipeline's id is the one the subscription's commit mode names.
        b.include(enrich)
            .out_reply(EosPublish::new("enrichment-1"))
            .transform(EosReplies);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(KafkaBroker::new(["localhost:9092"]))
}

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

per_partition делает идентификатор основой: издатели invoice получают идентификаторы invoicing-1-p0, invoicing-1-p1 и так далее.

EosPublish::new("enrichment-1") - это конвейер, и его идентификатор тот же, что называет режим фиксации подписки. EosReplies переносит исходную позицию каждой доставки на её ответ. По ней конвейер и сопоставляет одно с другим. Без него первый ответ завершается ошибкой, а не уходит мимо транзакции.

Окно фиксации​

Конвейер собирает ответы в окно и фиксирует его каждые commit_interval (по умолчанию 100 мс). В конце окна он ждёт, пока завершится каждая доставка, которая в него публиковала, добавляет их позиции и метаданные группы в транзакцию и фиксирует её. Поэтому ответ становится видимым не раньше чем через один интервал фиксации после того, как обработчик его вернул.

Неудачная публикация, неудачная фиксация или доставка, которая не завершилась вовремя, прерывают окно. Тогда консьюмеры перематываются к последним зафиксированным смещениям, и всё окно обрабатывается заново в новой транзакции. В зафиксированном выходе каждый ответ всё равно встречается один раз.

Конвейер ограничивает, что обработчики могут возвращать и как они читают:

  • retry() из обработчика держит окно открытым до истечения срока транзакции, 30 секунд, если transaction_timeout не задаёт другой, а затем окно читается заново.
  • retry_after не место в обработчике конвейера: его отложенная копия публикуется вне окна и нарушает соответствие записей и смещений.
  • KafkaPartitions не сочетается с Commit::Transactional, и такая подписка не запускается: конвейер фиксирует через группу консьюмеров, а читатель закреплённых партиций в группе не участвует.