Testing

The testing feature ships KafkaTestBroker, an in-process stand-in for Kafka: the same handlers and descriptors, no cluster. Enable it as a dev-dependency only:

[dev-dependencies]
ruststream-rdkafka = { version = "0.5", features = ["testing"] }

Never enable this feature in production builds.

#[subscriber(KafkaTopic::new("payments"))]
async fn accept(payment: &Payment) -> HandlerResult {
    if payment.amount == 0 {
        return HandlerResult::drop();
    }
    HandlerResult::Ack
}

TestApp::publish waits until the handlers settle before returning, so assertions read finished state - no sleeps:

let app = RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
    KafkaTestBroker::new(),
    |b| {
        b.include(accept);
    },
);
let tb = TestApp::start(app).await.expect("start");

tb.broker::<KafkaTestBroker>()
    .publish("payments", &Payment { amount: 100 })
    .await
    .expect("publish drives the handler to quiescence");

tb.broker::<KafkaTestBroker>()
    .subscriber("payments")
    .assert_called_once()
    .with(&Payment { amount: 100 })
    .settled(HandlerResult::Ack);

tb.shutdown().await.expect("shutdown");

What the test broker does not simulate

The in-process broker implements the core routing contract: exact topic-name fanout, settlement, headers, and the partition-key header. It deliberately does not simulate Kafka itself - consumer groups, partitions, committed positions, start offsets, rebalancing, and retention are transport behavior. nack(true) redelivers immediately in-process, while the real transport redelivers from the committed position on the next fetch.

Exercise the real semantics against a live cluster:

just brokers-up
KAFKA_TEST_URL=127.0.0.1:9092 cargo test --workspace --all-features -- --test-threads=1

The crate's own suites follow the same split: tests/testing_core.rs drives the in-process broker, and tests/integration_rdkafka.rs plus the conformance lifecycle run only when KAFKA_TEST_URL is set.