Apache Kafka

If you already have a Kafka cluster or need streaming semantics — replayable log, partition-key ordering, consumer-lag autoscaling — this backend fits naturally. Best for high-throughput pipelines where log retention and replay matter.

What you need

A Kafka cluster in KRaft mode or with Zookeeper. For local dev:

docker run --rm -p 9092:9092 confluentinc/cp-kafka:latest

The integration tests use testcontainers with the Apache Kafka module, so any runnable example also spins up a container automatically.

Install

cargo add shove --features kafka

For TLS + SASL (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512):

cargo add shove --features kafka-ssl

For AWS MSK with IAM authentication:

cargo add shove --features kafka-msk-iam

Connect

    let broker = Broker::<Kafka>::new(KafkaConfig::new(&bootstrap)).await?;

KafkaConfig::new takes a bootstrap server address (e.g. localhost:9092). TLS and SASL are configured via KafkaTls and KafkaSasl on the config — available when the kafka-ssl feature is enabled.

Connecting to AWS MSK

The kafka-msk-iam feature adds IAM-based authentication for Amazon MSK clusters. It pulls in aws-config, aws-credential-types, and aws-msk-iam-sasl-signer. Use it alongside kafka-ssl:

cargo add shove --features kafka-ssl,kafka-msk-iam

MSK IAM clusters listen on port 9098 (SASL/IAM). SCRAM/PLAIN clusters use port 9096.

Minimal setup

MSK brokers use publicly-signed ACM certificates. KafkaTls::default() is correct — the OS trust store handles validation. No custom CA path is needed.

use shove::kafka::{KafkaConfig, KafkaSasl, KafkaTls};
 
let config = KafkaConfig::new("b-1.cluster.amazonaws.com:9098,b-2.cluster.amazonaws.com:9098")
    .with_tls(KafkaTls::default())
    .with_sasl(KafkaSasl::msk_iam("eu-west-2"));

KafkaSasl::msk_iam(region) resolves credentials from the standard AWS provider chain: environment variables, shared credentials file, EC2 instance metadata (IMDS), EKS pod identity / IRSA, and SSO. No explicit credential configuration is needed in most deployment environments.

For a non-default named profile, use KafkaSasl::msk_iam_with_profile:

let config = KafkaConfig::new(brokers)
    .with_tls(KafkaTls::default())
    .with_sasl(KafkaSasl::msk_iam_with_profile("eu-west-2", "production"));

The OAUTHBEARER mechanism and SASL_SSL security protocol are set automatically. Do not set sasl.mechanism or security.protocol manually, and do not set sasl.oauthbearer.config — token rotation is handled automatically by the library.

See examples/kafka/msk_iam.rs for a runnable walkthrough.

Declare topology

    broker.topology().declare::<OrderTopic>().await?;

topology().declare::<T>() creates the Kafka topics for the main topic, hold topics, and DLQ. Idempotent — safe to call on every startup.

Publish

    let publisher = broker.publisher().await?;
    for i in 0..3 {
        publisher
            .publish::<OrderTopic>(&OrderCreated {
                order_id: format!("ORD-{i}"),
                amount: 99.99 + i as f64,
            })
            .await?;
        println!("Published order ORD-{i}");
    }

publisher().await? returns a Publisher<Kafka>. Messages are produced via rdkafka. The message key is derived from the topic's partition-key logic (or from T::sequence_key() for sequenced topics).

Producer tuning

shove pins the correctness-critical producer settings (acks=all, enable.idempotence=true) and keeps them non-configurable. Idempotence caps max.in.flight.requests.per.connection at 5, so sustained throughput is bounded by messages per request — i.e. by batching. librdkafka's defaults (linger.ms=5, no compression) favor low latency; high-rate pipelines can raise the throughput ceiling with three opt-in knobs on KafkaConfig:

use shove::kafka::{KafkaCompression, KafkaConfig};
 
let config = KafkaConfig::new(brokers)
    .with_producer_compression(KafkaCompression::Lz4)
    .with_producer_linger_ms(25)
    .with_producer_batch_size(500_000);

• with_producer_compression(KafkaCompression) — compression.type (None, Gzip, Snappy, Lz4, Zstd). Compresses each batch on the client, cutting producer→broker bytes as well as broker storage and replication traffic.
• with_producer_linger_ms(u32) — linger.ms, how long the producer waits to accumulate a batch. Higher values trade a little latency for materially larger (and better-compressed) batches. Must stay below the pinned message.timeout.ms (5000): linger time counts toward the message timeout, so values at or above it would expire every publish. Rejected at connect.
• with_producer_batch_size(u32) — batch.size, maximum bytes accumulated per batch (1..=i32::MAX, checked at connect). librdkafka caps the effective batch at min(batch.size, message.max.bytes), and message.max.bytes (default 1 MB) is not exposed — so this knob can lower the cap but not raise it above 1 MB.

Each knob left unset keeps librdkafka's default, so an untuned config behaves exactly as before. acks, enable.idempotence, and max.in.flight.requests.per.connection are deliberately not exposed.

The client holds a single producer, shared by publisher() and by the consumer's retry, defer, and DLQ republishes — a long linger also delays those republishes (and the offset commits gated on them for sequenced topics), so keep linger modest when consumers republish on the same client.

Consume

    let mut group = broker.consumer_group();
    group
        .register::<OrderTopic, _>(
            ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=1)),
            || OrderHandler,
        )
        .await?;
 
    // Stop after 3 s for demo purposes, or on ctrl-c.
    let outcome = group
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(3)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(10),
        )
        .await;

consumer_group() registers a native Kafka consumer group. KafkaConsumerGroupConfig::new(min..=max) sets the autoscale bounds. Kafka handles partition assignment and rebalance automatically when the group membership changes.

Group configuration

use shove::kafka::{KafkaAutoOffsetReset, KafkaConsumerGroupConfig};

let cfg = KafkaConsumerGroupConfig::new(1..=8)
    .with_prefetch_count(20)
    .with_max_retries(5)
    .with_handler_timeout(Duration::from_secs(30))
    .with_concurrent_processing(true)
    .with_group_id("billing-orders-consumer")          // override the default `{queue}-consumer`
    .with_auto_offset_reset(KafkaAutoOffsetReset::Latest);

• with_prefetch_count(u16) — librdkafka in-flight cap per consumer task. Default 10.
• with_max_retries(u32) — retry budget before dead-lettering. Default 10.
• with_handler_timeout(Duration) — per-message wall-clock deadline. Default 30s (see Handlers & Context).
• with_concurrent_processing(bool) — dispatch each fetched message to its own tokio task (rejected for sequenced topics). Default false.
• with_group_id(impl Into<String>) — override the broker-side consumer group ID. Defaults to "{queue}-consumer". Set this when two independent services consume the same topic and must each receive every message (fan-out) — otherwise they share a group and compete for partitions. Prefer .for_consumer_group(...) on the topology (below), which sets the group and the DLQ/hold-queue names together.
• with_auto_offset_reset(KafkaAutoOffsetReset) — Earliest (default, replay history), Latest (tail-only), or None (refuse silent replay/skip on a fresh group).

Fan-out — a second reader on the same topic

A bare with_group_id splits the group but not the retry chain: both readers still derive {queue}-dlq and {queue}-hold-*, so each drains the other's dead and held messages. Declaring the second reader's topology with for_consumer_group splits both:

TopologyBuilder::new("order-settlement")
    .for_consumer_group("settlement-audit")
    .hold_queue(Duration::from_secs(5))
    .dlq()                                   // order-settlement-settlement-audit-dlq
    .build()

The resolved group IDs follow the topology:

Precedence is: an explicit with_group_id (on either KafkaConsumerGroupConfig or ConsumerOptions::<Kafka>) > the topology's fan-out group > the {queue}-consumer default. The explicit override staying on top means adding for_consumer_group to a topology cannot move an already-deployed consumer off the group it holds committed offsets under. The autoscaler resolves the same group ID as the broker-side consumer in every case, so lag is read from the group that is actually committing.

Re-anchoring a group (seek to tail / head / timestamp)

auto.offset.reset only decides where a group starts when it has no usable committed offset. Once the group has committed, the setting is inert — which is why "just seek to the tail" so often turns into minting a throwaway group ID (orders-v2, orders-20260812, …). That works, but it strands the old group's offsets and its lag metrics forever, and the generation suffix becomes a permanent piece of config nobody dares remove.

reset_consumer_group_offsets rewrites the group's committed offsets in place — the library-side equivalent of kafka-consumer-groups.sh --reset-offsets --execute:

use shove::kafka::{KafkaConsumerGroupConfig, KafkaOffsetReset};

let config = KafkaConsumerGroupConfig::new(1..=4);

// Operator-initiated: re-anchor at the tail before the consumers start.
if std::env::var("PRICES_SEEK_TO_TAIL").is_ok() {
    let report = broker
        .reset_consumer_group_offsets::<Prices>(&config, KafkaOffsetReset::Latest)
        .await?;
    tracing::warn!(?report, "re-anchored the prices group at the tail");
}

let mut group = broker.consumer_group();
group
    .register::<Prices, _>(ConsumerGroupConfig::new(config), || Handler)
    .await?;

• KafkaOffsetReset::Latest — every partition's high watermark. The seek-to-tail case: a latest-value sink that must serve fresh data now rather than after crawling days of backlog.
• KafkaOffsetReset::Earliest — every partition's low watermark: replay all retained history.
• KafkaOffsetReset::Timestamp(ms) — the first record at or after that point, in milliseconds since the Unix epoch (the same unit as --to-datetime). Partitions with no record at or after it re-anchor at their high watermark.

The group ID is resolved from config and the topology exactly as register would resolve it — following the same precedence as the fan-out table above, plus the -fifo suffix for a sequenced topic — so the offsets rewritten are the ones the consumers will actually read.

The group must be inactive. Kafka only accepts an offset reset while the group has no live members; with consumers running the call returns ShoveError::Validation naming the active member count, and the broker enforces the same rule independently. Re-anchor at process start, before the group is registered. A group does not go inactive the instant its consumers stop — the coordinator drops each member as its LeaveGroup lands — so a reset issued immediately after run_until_timeout returns may need a brief retry.

The returned KafkaOffsetResetReport carries one entry per partition with the previous and new offsets (and delta(), positive for records skipped, negative for history replayed). It is the only record of where the group was before it moved, so log it. is_noop() reports that every partition already sat at its target.

Kafka is the only backend with this API: it is the only one shove supports where a group's read position is a broker-side committed offset an operator can rewrite. Redis Streams' XGROUP SETID is the nearest equivalent and is not yet exposed.

Replication factor

Auto-created topics get replication factor 1 by default, which is fine for single-broker dev but unsafe in production. Set a cluster-wide default on the registry:

let mut group = broker
    .consumer_group()
    .with_default_replication_factor(3);                // applied to every auto-created topic

Or set it per-declaration on the topology declarer:

broker
    .topology()
    .with_replication_factor(3)
    .declare::<Orders>()
    .await?;

create_topic is idempotent and will not lower an existing topic's replication. Pre-creating topics out of band (Terraform, MSK console) is also fine — the declarer is a no-op when the topic already exists.

Sequenced delivery

Messages for the same key stay in order. Kafka uses partition-key routing: T::sequence_key() becomes the Kafka message key, so messages with the same key always land on the same partition. A single consumer handles each partition at a time, guaranteeing that messages for the same key are never processed concurrently.

Ordering is partition-scoped: two messages with different keys may land on different partitions and be processed concurrently. The partition count is fixed at topic creation time and caps the maximum degree of parallelism across the consumer group.

See Sequenced Topics for the full ordering model, and Sequenced example for a runnable walkthrough.

Consumer groups + autoscaling

Kafka consumer groups are native — broker.consumer_group() creates a standard Kafka consumer group. Call register to associate a topic with handlers and bounds:

use shove::kafka::KafkaConsumerGroupConfig;
use shove::{Broker, ConsumerGroupConfig, Kafka};

let mut group = broker.consumer_group();
group
    .register::<OrderTopic, _>(
        ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=4)),
        || MyHandler,
    )
    .await?;

The autoscaler measures consumer lag (the offset gap between the latest produced message and the latest committed offset) and adjusts the number of active consumers within the min..=max range.

Note: consumer-group rebalances occur during scale-up and scale-down events. Rebalances cause a brief delivery pause while Kafka reassigns partitions. Plan for this when setting autoscale bounds and drain timeouts.

See the Basic example for a full runnable walkthrough.

Offset commit semantics

Outcome::Ack commits the offset immediately after the handler returns. Outcome::Retry and Outcome::Defer defer the offset commit until the delayed republish to the hold topic has been acknowledged by the broker. This closes the publish-then-commit race: a broker crash or process kill between handler return and republish redelivers the message on restart rather than silently dropping it.

On shutdown the consumer drains the in-flight republish queue before exiting the poll loop, so pending Retry/Defer messages are persisted even if the shutdown signal arrives mid-batch.

Outcome::Reject routes to the DLQ (if configured) and commits the offset; the original is not redelivered. Messages that exhaust max_retries follow the same DLQ-then-commit path.

Schema Registry

The kafka-schema-registry feature adds Confluent Schema Registry support for Kafka consumers (decode) and producers (encode). On the consume side, each incoming message is unwrapped from the Confluent wire frame (magic byte 0x00 + 4-byte big-endian schema id; for Protobuf, also the message-index array), the schema id is resolved against the registry (cached in memory, with single-flight deduplication of concurrent cold misses and a configurable negative-TTL for permanent failures), and the inner payload is decoded via the topic's existing Codec (JsonCodec or ProtobufCodec). On the produce side, an opt-in publisher wraps each encoded payload in the same wire frame — see Producer-side encoding.

The feature works with the Confluent Schema Registry and with Redpanda's built-in Schema Registry, which exposes the same Confluent-compatible REST API and wire format. The e2e test suite validates both paths.

Install

cargo add shove --features kafka-schema-registry

Build the registry client

use std::time::Duration;
use shove::schema_registry::{SchemaRegistry, SchemaRegistryAuth};

let registry = SchemaRegistry::builder("https://schema-registry:8081")
    .auth(SchemaRegistryAuth::Basic {
        user: "sr-user".into(),
        pass: "sr-pass".into(),
    })
    .timeout(Duration::from_secs(3))
    .max_retries(2)
    .negative_cache_ttl(Duration::from_secs(60))
    .build();

SchemaRegistry::builder(url) accepts any http:// or https:// base URL, with or without user:pass@ userinfo. Auth options are:

• SchemaRegistryAuth::None — no authentication (default)
• SchemaRegistryAuth::Bearer(token) — Authorization: Bearer <token>
• SchemaRegistryAuth::Basic { user, pass } — HTTP Basic auth

Credentials require TLS. Configuring any credential against a base URL that is not https:// makes build() panic, because the secret would be sent in cleartext on every schema fetch. Three things count as a credential:

• any SchemaRegistryAuth other than None;
• user:pass@ userinfo in the base URL — the HTTP client lifts it out and replays it as an Authorization: Basic header, so it is a credential even with SchemaRegistryAuth::None;
• any header set with .header(...), whose value is assumed to be a secret. Use .non_secret_header(...) for one that is not.

The base URL is parsed before it is judged, so alternate spellings of a plaintext scheme — http:registry:8081, http:/registry:8081 — are refused too rather than slipping past a prefix check. A base URL that does not parse is treated as plaintext.

A credential-bearing client also stops following redirects. A schema registry has no legitimate reason to redirect, and a 302 would otherwise replay a custom secret header to whatever host and scheme the Location names. An unauthenticated client is unaffected.

An unauthenticated http:// registry is unaffected. For a registry that is genuinely unreachable from an untrusted network — a local development stack — opt in explicitly:

use shove::schema_registry::{SchemaRegistry, SchemaRegistryAuth};

let registry = SchemaRegistry::builder("http://localhost:8081")
    .auth(SchemaRegistryAuth::Bearer("dev-token".into()))
    .allow_plaintext_credentials()
    .build();

The returned value is an Arc<SchemaRegistry>. Clone it to share the same schema cache across multiple consumers or a consumer group.

Per-consumer configuration

use std::sync::Arc;
use shove::schema_registry::{SchemaEnforcement, SchemaRegistry, SchemaRegistryAuth};
use shove::consumer::ConsumerOptions;
use shove::markers::Kafka;

let registry = SchemaRegistry::builder("http://schema-registry:8081").build();

let opts = ConsumerOptions::<Kafka>::new()
    .with_schema_registry(Arc::clone(&registry))
    .with_schema_enforcement(SchemaEnforcement::Enforce)
    .accept_schema_subjects(["orders-value"]);

Per-consumer-group configuration

Attaching the registry on a KafkaConsumerGroupConfig shares the Arc — and therefore the same in-memory schema cache — across every consumer spawned in the autoscaling group:

use std::sync::Arc;
use shove::kafka::{KafkaConsumerGroupConfig};
use shove::schema_registry::{SchemaEnforcement, SchemaRegistry};

let registry = SchemaRegistry::builder("http://schema-registry:8081").build();

let cfg = KafkaConsumerGroupConfig::new(1..=8)
    .with_schema_registry(Arc::clone(&registry))
    .with_schema_enforcement(SchemaEnforcement::Enforce)
    .accept_schema_subjects(["orders-value"]);

Enforcement modes

with_schema_enforcement controls what happens when a message's registered subject is not in the accepted set:

• SchemaEnforcement::Enforce (default) — the message is routed to the DLQ with death reason schema_validation_failed. Choose this for production where a subject mismatch is a producer misconfiguration.
• SchemaEnforcement::Permissive — the mismatch is logged and counted, and the message is decoded anyway. Use this during migration windows or when multiple producers share a topic with different subject conventions.

Accepted subjects

accept_schema_subjects([...]) pins the set of Confluent schema subjects that are accepted for a consumer. When not called, the default is derived from the queue name using the Confluent TopicNameStrategy: "{queue}-value". For a topic named orders, that resolves to orders-value.

Producer-side encoding

The same feature lets a publisher emit Confluent-framed messages, symmetric with how the consumer is configured. Attach a SchemaRegistry to a KafkaPublisherConfig and obtain the publisher with Broker::publisher_with; the publisher then wraps each codec-encoded payload in the Confluent wire frame using the latest registered schema id for the subject. Like the consumer side, framing is a publisher-layer concern — it is not baked into the topic's Codec.

use std::sync::Arc;
use shove::kafka::KafkaPublisherConfig;
use shove::schema_registry::SchemaRegistry;

let registry = SchemaRegistry::builder("http://schema-registry:8081").build();

// `broker` is a `Broker<Kafka>`.
let publisher = broker
    .publisher_with(KafkaPublisherConfig::new().with_schema_registry(Arc::clone(&registry)))
    .await?;

// `publisher.publish::<OrdersTopic>(&order).await?` now emits SR-framed bytes.

Details:

• Subject — defaults to the Confluent TopicNameStrategy "{topic}-value"; override with KafkaPublisherConfig::with_subject("...").
• Schema id — resolved once per subject via GET /subjects/{subject}/versions/latest and cached. shove carries no schema text; the subject must already be registered.
• Codecs — framing applies only to JsonCodec and ProtobufCodec topics; other codecs are published unframed. For Protobuf the message-index is [0] (the first message type).
• Opt-in — a publisher built with the plain Broker::publisher() does no framing, so existing publishers are byte-for-byte unchanged.

Codec requirement

Registry decoding is only supported for topics using a JsonCodec or ProtobufCodec. A message arriving on a topic whose codec is neither JSON nor Protobuf is routed to the DLQ with reason schema_unsupported_codec.

Redpanda compatibility

The Confluent wire format (magic byte + schema id) and REST API endpoints used by kafka-schema-registry are fully compatible with Redpanda's built-in Schema Registry. No additional configuration is needed — point SchemaRegistry::builder at the Redpanda Schema Registry URL.

DLQ consumer caveat

Metrics

The Kafka backend emits shove_messages_failed_total with the following reason labels (see Observability):

• oversize — payload exceeds max_message_size before deserialization.
• deserialize — JSON/codec decode failed; routed to DLQ.
• timeout — handler exceeded handler_timeout; retried.
• max_retries_exceeded — retry budget exhausted; routed to DLQ.
• rejected — handler returned Outcome::Reject; routed to DLQ.
• schema_validation_failed — schema subject not in the accepted set under Enforce mode; routed to DLQ. Requires kafka-schema-registry.
• schema_unsupported_codec — registry decoding requested but the topic uses a codec other than JSON or Protobuf; routed to DLQ. Requires kafka-schema-registry.

shove_backend_errors_total carries the backend="kafka" label for connection drops, produce failures, and consume-stream errors.

Gotchas

• Replication factor defaults to 1 — safe for single-broker dev, unsafe in production. Set with_default_replication_factor(3) on the registry (or with_replication_factor on the topology declarer) before the first declaration. The default exists so the no-config dev path works; the constant lives in kafka::constants::DEFAULT_REPLICATION.
• Partition count is set at topic creation time and is rarely changed after the fact. Choose carefully — partition count caps the maximum degree of parallelism a consumer group can achieve (one consumer per partition maximum).
• Consumer-group rebalances during scale events cause brief delivery pauses. The larger the group and the more topics, the longer the rebalance. shove pins session.timeout.ms to 10 seconds and max.poll.interval.ms to 5 minutes (constants in kafka::constants); neither is settable via KafkaConfig. If handlers are slow, raise the handler timeout with with_handler_timeout (per group) or with_default_handler_timeout (registry-wide) instead.
• On Windows, the cmake-build feature is required for rdkafka (the underlying C library). This is handled automatically via the [target.'cfg(windows)'.dependencies] entry in Cargo.toml.
• SASL and TLS require the kafka-ssl feature. SCRAM-SHA-256 and SCRAM-SHA-512 are supported out of the box. GSSAPI/Kerberos requires downstream activation of rdkafka/gssapi. For AWS MSK IAM, use the kafka-msk-iam feature instead.
• librdkafka system dependencies on Debian/Ubuntu CI runners: libsasl2-dev is required when using kafka-ssl. Add it to your CI runner setup (the shove ci.yml already does this).

Examples

• Basic — publish/consume round trip with hold queues and DLQ
• Sequenced — partition-key ordering
• Audited Consumer — MessageHandlerExt::audited wrapping
• Stress — throughput benchmarking
• MSK IAM — IAM authentication against Amazon MSK
• Schema Registry — Confluent Schema Registry decode with subject enforcement

See also

• Liveness Probes — wire Broker::ping into a k8s health endpoint.
• Broadcast — per-instance fan-out via a groupless assign() at the tail, leaving __consumer_offsets untouched.