跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

测试 Kafka 服务

测试用 TestApp 运行服务自己的应用，连接的是进程内 Kafka 或者集群。

被测服务负责银行卡扣款。 付款到达 card-payments，它的记录键就是要扣款的账户。 处理器在 card-receipts 上以一张收据作为回复。 没有键的付款被丢弃。 超出银行即时限额的付款会被暂缓：处理器用 retry_after 要求一分钟后再投递它。 经过三次投递后，付款进入死信主题 card-payments.review。 应用由 service(broker) 构建，所以测试可以选择 Broker，而 main 使用生产环境的 Broker。 测试需要在 [dev-dependencies] 中启用 ruststream-rdkafka 的 testing feature。

带键的付款​

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。 测试从不连接引导地址。 测试按生产类型访问 Broker：tb.broker::<KafkaBroker>()。 Kafka 的记录键以 PARTITION_KEY_HEADER 消息头传递，所以测试用 with_headers 设置它。 Broker 把这个消息头变成记录的原生键，处理器再用 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 会读取的配置：Broker 的 config 和 producer_config、每个描述符的选项，以及 librdkafka 的默认值。 librdkafka 拒绝的属性会以同样的方式让测试启动失败。 它模拟的是 Kafka，而不是一个队列：

  • 每个消费者组对一条记录只读取一次，每个组读取自己的一份。
  • 提交模式决定提交什么：Commit::Tracked 下的 retry 会回来，Commit::Auto 下则报告自己不受支持。
  • 记录键决定记录放在哪里，并回到处理器。
  • seek 把订阅移到某个偏移量、日志的任一端，或者某个时间戳。
  • 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。 它的截止时间是一次发布等待其引发的反应完成结算的最长时间；真实集群需要比进程内默认值更长的时间。 测试用 live::url（示例自带的辅助函数）从 KAFKA_TEST_URL 读取地址，未设置时跳过。 它加入自己的消费者组，所以不会读到之前运行留下的任何东西。 它先创建 card-payments：订阅在主题已有的分区上打开，之后才出现的主题要从 auto.offset.reset 指定的位置读取。 断言与进程内的完全相同。

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