Skip to main content

Testing a Kafka service

A test runs the service's own app with TestApp, against an in-process Kafka or against a cluster.

The service under test charges card payments. A payment arrives on card-payments, and its record key is the account to charge. The handler replies with a receipt on card-receipts. A payment without a key is dropped. A payment over the bank's instant limit is held: the handler asks for it again in a minute with retry_after. After three deliveries a payment goes to card-payments.review, the dead-letter topic. The app is built by service(broker), so a test chooses the broker and main uses the production one. The tests need the testing feature of ruststream-rdkafka in [dev-dependencies].

A keyed payment​

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 connects KafkaBroker to an in-process Kafka. The bootstrap address is never dialled. The test addresses the broker by its production type, tb.broker::<KafkaBroker>(). A Kafka record key travels as the PARTITION_KEY_HEADER header, so the test sets it with with_headers. The broker turns that header into the record's native key, and the handler reads it back with Ctx<Key>. publish returns once the handler and its reply have settled. So the next lines assert on finished state: how the subscriber settled and what reached card-receipts. The body takes the started app as an argument, so the live test below runs it too.

A payment without a key​

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 checks the outcome the handler returned. assert_not_called checks that no receipt was published.

A delay, without waiting for it​

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(())
}

The test starts on a paused clock, and advance moves it forward. A minute of HOLD costs nothing. Kafka cannot hold a record back, so retry_after publishes a copy of the record to its topic when the delay ends. The copy carries the retry count in a header, and published shows it on card-payments next to the test's own record. The third delivery is the last one the mount site allows, so the record goes to card-payments.review as it arrived. Each held delivery is recorded as Outcome::Nack.

What the in-process Kafka models​

The in-process Kafka reads the configuration librdkafka would read: the broker's config and producer_config, each descriptor's options, and librdkafka's defaults. A property librdkafka refuses fails the test's start the same way. It models Kafka rather than a queue:

  • Each consumer group reads a record once, and every group reads its own copy.
  • The commit mode decides what is committed: a retry under Commit::Tracked comes back, and under Commit::Auto it reports itself unsupported.
  • A record key places the record and comes back to the handler.
  • A seek moves a subscription to an offset, either end of the log, or a timestamp.
  • A read_committed reader sees a transaction's records when it commits and never after an abort.

Every topic has one partition there. Several partitions, cooperative rebalancing, retention and the timing of a real group join need a cluster.

The same test on a cluster​

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 connects the same app to a running Kafka. Its deadline is the longest a publish waits for the reaction to settle; a real cluster needs more than the in-process default. The test reads the address from KAFKA_TEST_URL with live::url, a helper of the example's own, and skips when it is unset. It joins a consumer group of its own, so nothing an earlier run left behind is read. It creates card-payments first: a subscription opens on the partitions a topic has, and a topic that appears later is read from where auto.offset.reset puts it. The assertions are the in-process ones, unchanged.

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