Live

TestApp::start_live runs the same test body against a running Kafka, for what only a cluster has.

It connects the broker through its ordinary connect, so startup probes the cluster and every group joins it for real. The body is the one an in-process test runs; only the start call differs. A publish, settle and advance wait on the real clock until every subscription has handled what was published to it. The wait lasts ten seconds at most, and start_live_within sets another bound. A consumer group takes a few seconds to form on a cluster, so a Kafka test gives it thirty.

The live mode is for what the in-process cluster leaves out: a topic created with several partitions, the cooperative assignment that keeps partitions sticky, retention, and the timing of a group join. A cluster keeps what earlier runs left in it. Each run reads in a consumer group of its own, and creates its topics before the app starts: a group assigned a topic created after it subscribed reads from the end of the log, past the record that created the topic.

In the example, live::url returns the cluster's address from KAFKA_TEST_URL, and the test returns early without one, so cargo test passes on a machine with no cluster. A job that stood a cluster up sets RUSTSTREAM_REQUIRE_LIVE=1, and a missing address then fails the test instead of skipping it. The repository's stand is docker-compose.test.yml, which just brokers-up starts.

Example

The two tests share one body: one in process, one against the cluster at KAFKA_TEST_URL.

Macros
Manual

examples/kafka_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<KafkaBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .subscriber("testing-live-payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app("kafka:9092", "payments-svc")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(servers) = live::url("KAFKA_TEST_URL") else {
        return Ok(());
    };
    ensure_topic(&servers, "testing-live-payments").await?;
    // A group of this run's own, so nothing an earlier run committed is read.
    let group = format!("payments-{}", process::id());
    // A group forms on a cluster in seconds, more than the default ten-second bound allows for.
    let tb = TestApp::start_live_within(app(&servers, &group), Duration::from_secs(30)).await?;
    accepts_a_payment(tb).await
}

examples/manual/kafka_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<KafkaBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<KafkaBroker>()
        .subscriber("testing-live-payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app("kafka:9092", "payments-svc")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(servers) = live::url("KAFKA_TEST_URL") else {
        return Ok(());
    };
    ensure_topic(&servers, "testing-live-payments").await?;
    // A group of this run's own, so nothing an earlier run committed is read.
    let group = format!("payments-{}", process::id());
    // A group forms on a cluster in seconds, more than the default ten-second bound allows for.
    let tb = TestApp::start_live_within(app(&servers, &group), Duration::from_secs(30)).await?;
    accepts_a_payment(tb).await
}

Example​

Example