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

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

Тестирование сервиса на Kafka

Тест запускает собственное приложение сервиса через TestApp на Kafka внутри процесса или на кластере.

Тестируемый сервис списывает платежи по картам. Платёж приходит в тему card-payments, и ключ его записи - счёт для списания. Обработчик отвечает квитанцией в тему card-receipts. Платёж без ключа отбрасывается. Платёж сверх мгновенного лимита банка откладывается: обработчик просит доставить его снова через минуту с retry_after. После трёх доставок платёж уходит в card-payments.review, тему dead-letter. Приложение строит service(broker), поэтому тест выбирает брокер сам, а main берёт рабочий. Тестам нужна фича testing крейта ruststream-rdkafka в [dev-dependencies].

Платёж с ключом​

src/main.rs
#[tokio::test]
async fn a_keyed_payment_gets_a_receipt() -> Result<(), Box<dyn Error>> {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("payments-svc");
    let tb = TestApp::start(service(broker)).await?;
    a_receipt_names_the_account_of_the_key(tb).await
}

async fn a_receipt_names_the_account_of_the_key(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    let mut key = HeaderMap::new();
    key.insert(PARTITION_KEY_HEADER, "acct-7");
    tb.broker::<KafkaBroker>()
        .message(&Payment {
            id: 1,
            amount_cents: 2_500,
        })
        .with_headers(key)
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<KafkaBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt {
            payment: 1,
            account: "acct-7".to_owned(),
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start подключает KafkaBroker к Kafka внутри процесса. Соединение по начальному адресу не устанавливается. Тест обращается к брокеру по его рабочему типу: tb.broker::<KafkaBroker>(). Ключ записи Kafka передаётся заголовком PARTITION_KEY_HEADER, поэтому тест задаёт его через with_headers. Брокер превращает этот заголовок в настоящий ключ записи Kafka, а обработчик читает его обратно через Ctx<Key>. publish возвращает управление, когда обработчик и его ответ завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписчик завершил доставку и что дошло до card-receipts. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на кластере ниже.

Платёж без ключа​

src/main.rs
#[tokio::test]
async fn a_payment_without_an_account_is_dropped() -> Result<(), Box<dyn Error>> {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("payments-svc");
    let tb = TestApp::start(service(broker)).await?;

    tb.broker::<KafkaBroker>()
        .message(&Payment {
            id: 2,
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<KafkaBroker>()
        .published::<Receipt>("card-receipts")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled проверяет исход, который вернул обработчик. assert_not_called проверяет, что квитанция не опубликована.

Задержка без ожидания​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_payment_the_bank_keeps_holding_goes_to_review() -> Result<(), Box<dyn Error>> {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("payments-svc");
    let tb = TestApp::start(service(broker)).await?;
    let held = Payment {
        id: 3,
        amount_cents: 250_000,
    };
    let mut key = HeaderMap::new();
    key.insert(PARTITION_KEY_HEADER, "acct-7");

    tb.broker::<KafkaBroker>()
        .message(&held)
        .with_headers(key)
        .publish()
        .await?;
    tb.broker::<KafkaBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The delay is a copy of the record, published back to the topic after `HOLD`.
    tb.advance(HOLD).await?;
    tb.broker::<KafkaBroker>()
        .published::<Payment>("card-payments")
        .assert_called(2)
        .with(&held)
        .with_header(RETRY_COUNT_HEADER, "1");

    // The third delivery is the last one: it goes to the review topic as it arrived.
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<KafkaBroker>()
            .subscriber("card-payments")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<KafkaBroker>()
        .published::<Payment>("card-payments.review")
        .assert_called_once()
        .with(&held);
    tb.shutdown().await?;
    Ok(())
}

Тест начинается с остановленными часами, а advance сдвигает их вперёд. Минута HOLD не стоит ничего. Kafka не умеет придерживать запись, поэтому по окончании задержки retry_after публикует копию записи в её тему. Копия несёт число попыток в заголовке, и published показывает её в card-payments рядом с собственной записью теста. Третья доставка - последняя, которую допускает точка монтирования, поэтому запись уходит в card-payments.review в том виде, в каком пришла. Каждая отложенная доставка записывается как Outcome::Nack.

Что моделирует Kafka внутри процесса​

Kafka внутри процесса читает ту же конфигурацию, что и librdkafka: config и producer_config брокера, параметры каждого дескриптора и значения librdkafka по умолчанию. Свойство, которое librdkafka отвергает, так же проваливает запуск теста. Она моделирует Kafka, а не очередь:

  • Каждая группа консьюмеров читает запись один раз, и каждая группа читает свою копию.
  • Режим фиксации решает, что фиксируется: retry при Commit::Tracked возвращает запись, а при Commit::Auto сообщает, что не поддерживается.
  • Ключ записи определяет её размещение и возвращается в обработчик.
  • Перемотка переносит подписку на смещение, на любой конец журнала или на отметку времени.
  • Читатель read_committed видит записи транзакции после её фиксации и никогда после прерывания.

У каждой темы там одна партиция. Для нескольких партиций, кооперативной перебалансировки, срока хранения и времени входа в настоящую группу нужен кластер.

Тот же тест на кластере​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_keyed_payment_gets_a_receipt_on_a_cluster() -> Result<(), Box<dyn Error>> {
    let Some(servers) = live::url("KAFKA_TEST_URL") else {
        return Ok(());
    };
    create_topic(&servers, "card-payments").await?;
    let group = format!("payments-svc-{}", process::id());
    let broker = KafkaBroker::new([servers]).default_group(group);
    let tb = TestApp::start_live_within(service(broker), Duration::from_secs(30)).await?;
    a_receipt_names_the_account_of_the_key(tb).await
}

TestApp::start_live_within подключает то же приложение к работающей Kafka. Его срок - самое долгое время, которое публикация ждёт завершения реакции. Настоящему кластеру нужно больше, чем значение по умолчанию внутри процесса. Тест читает адрес из KAFKA_TEST_URL через live::url, собственную вспомогательную функцию примера, и пропускается, если переменная не задана. Он входит в собственную группу консьюмеров, поэтому не читает ничего, что оставил прошлый прогон. Сначала он создаёт card-payments: подписка открывается на партициях, которые есть у темы, а тема, появившаяся позже, читается с того места, куда её ставит auto.offset.reset. Проверки те же, что и внутри процесса, без изменений.

cargo test
KAFKA_TEST_URL=localhost:9092 cargo test -- --test-threads=1