Datum

Appearance

Kafka ​

The datum-mq crate gives Datum a native Kafka connector: KafkaSource for at-least-once consuming with committable offsets, and KafkaSink for producing with delivery confirmation. The behavior reference is alpakka/Pekko Connectors Kafka: committableSource to downstream work to a committer, plus a plain producer.

sh
cargo add datum-mq

Native-only in v0.11.0

datum-mq has one Kafka client implementation: Datum's Tokio-native producer and consumer. There is no selectable FFI compatibility backend, no Kafka C client dependency, and no C/CMake build for Kafka itself.

The rustls TLS path uses the ring provider (Rust plus assembly), so the TLS/crypto path does not compile aws-lc-rs, aws-lc-sys, or a CMake-built C TLS library. This is a precise TLS/crypto claim, not a whole-crate C-free claim: Kafka Zstd compression still uses zstd-sys through the Rust zstd crate.

Native Kafka supports plaintext, TLS, SASL PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512. Native produce and fetch support every Kafka record-batch codec: none, gzip, snappy, lz4, and zstd. The native producer is idempotent by default (acks=all, enable.idempotence=true). Kafka transactions, consume-transform-produce EOS, SASL GSSAPI, and SASL OAUTHBEARER are not supported; see Deferred scope.

The consumer's shipped offset-commit guarantee is at-least-once. Native producer retries preserve the no-duplicate producer guarantee by default.

Consuming — the committable flow ​

KafkaSource::committable emits a SourceWithContext whose context is a KafkaOffset. You commit that offset only after your own checkpoint or side effect succeeds — that is what makes the flow at-least-once:

rust
use datum::SourceWithContext;
use datum_mq::{ConsumerRecord, KafkaControl, KafkaOffset};

// At-least-once consumer: each record carries a committable KafkaOffset as
// context. Datum-managed manual commits are the default (CommitPolicy::Manual).
let settings = KafkaConsumerSettings::new("127.0.0.1:9092", "orders-consumer")
    .with("auto.offset.reset", "earliest")
    .with_commit_policy(CommitPolicy::Manual)
    .with_backpressure(2_048, 4_096);

let source: SourceWithContext<ConsumerRecord, KafkaOffset, KafkaControl> =
    KafkaSource::committable(settings, Subscription::topics(["orders"]));

// Building the blueprint connects nothing; materialization starts the
// consumer group and its poll loop.
let _ = source;

Commit-after-checkpoint semantics ​

Committing is a call on the offset once downstream work is durable:

  • KafkaOffset::commit() — marks this offset processed and advances the commit.
  • KafkaOffset::mark_processed() — stages the offset without a Kafka commit, for batched commits: stage the earlier offsets in a contiguous run, then commit() the last one after the batch checkpoint succeeds.

Under the hood the source tracks a per-partition processed watermark and only ever commits the highest contiguous processed offset per partition — so an out-of-order commit() never advances the committed offset past a gap. Actual Kafka commits are flushed by count (commit_batch_size, default 10,000) or by interval (commit_interval, default 100 ms), and forced synchronously on partition revoke and on source close. Manual commits are the default (CommitPolicy::Manual); CommitPolicy::AutoKafka is rejected by the native client because Kafka would own progress independently of downstream durability. Use CommitPolicy::External when offsets are stored outside Kafka.

This is the broker-gated shape end to end — commit each offset inside the fold, after the record is accepted (adapted from crates/datum-mq/tests/kafka_integration.rs):

rust
// Requires a Kafka broker. Adapted from tests/kafka_integration.rs.
use datum::{Sink, Source, StreamError};
use datum_mq::{ConsumerRecord, KafkaConsumerSettings, KafkaOffset, KafkaSource, Subscription};

let settings = KafkaConsumerSettings::new("127.0.0.1:9092", "orders-consumer")
    .with("auto.offset.reset", "earliest");

let completion = KafkaSource::committable(settings, Subscription::topics(["orders"]))
    .as_source() // (record, offset) pairs
    .run_with(Sink::fold_result(
        0_u64,
        |count, (record, offset): (ConsumerRecord, KafkaOffset)| {
            let _dedupe_key = (&record.topic, record.partition, record.offset);
            // Do your durable, idempotent work here.
            offset.commit().map_err(StreamError::from)?; // commit AFTER the work
            Ok(count + 1)
        },
    ))?;

Because it is at-least-once, a crash between processing and commit replays those records. Make downstream work idempotent, or de-duplicate on (topic, partition, offset) — all available on KafkaOffset.

Throughput-oriented batches ​

For high-throughput consuming, KafkaSource::committable_payload_batches emits one KafkaPayloadBatch per Kafka poll batch — payloads are copied into one contiguous buffer, and the batch carries a single committable watermark per touched partition (KafkaPayloadBatch::commit()). This avoids one offset handle per record while keeping at-least-once commit ordering. It is the measured counterpart to per-record committable.

Backpressure and draining ​

The source paces Kafka fetches against downstream demand: when outstanding (emitted-but-not-committed) offsets cross high_watermark it pauses fetching, and resumes below low_watermark (with_backpressure(low, high), default 2,048 / 4,096). Backpressure is per partition, so a slow partition does not force unrelated partitions to stop scheduling fetches.

KafkaControl — the source's materialized value — is the Datum equivalent of alpakka's DrainingControl:

  • drain_and_shutdown(timeout) — stop new emission, wait for already-emitted offsets to be committed, then complete the source on its next pull. This is what a datum-agent job's drain hook calls before stopping the graph.
  • shutdown_now() — stop without waiting for outstanding commits.
  • metrics() — a KafkaMetricsSnapshot (assigned/revoked/lost partitions, rebalances, emitted, committed offsets, commit failures, paused, outstanding, high/committed watermark).

Native consumer ​

The native consumer decodes Kafka Fetch responses directly into Datum batches. One response can contain many partitions and record batches; the source turns it into ConsumerRecord values for per-record flows or into one KafkaPayloadBatch with one owned payload buffer and committable watermarks for batch flows.

KafkaSource::plain, KafkaSource::committable, and KafkaSource::committable_payload_batches all use the native consumer:

rust
use datum_mq::{KafkaConsumerSettings, KafkaSource, Subscription};

let settings = KafkaConsumerSettings::new("127.0.0.1:9092", "orders-consumer")
    .with("auto.offset.reset", "earliest");

let source = KafkaSource::committable_payload_batches(
    settings,
    Subscription::topics(["orders"]),
);

Security and protocol boundary ​

Security settingNative consumerNative producerNotes
security.protocol=plaintextSupportedSupportedNo transport security.
security.protocol=sslSupportedSupportedTokio/rustls, system roots plus optional CA PEM, SNI verification, optional mTLS.
security.protocol=sasl_plaintextSupportedSupportedTest-only; credentials and traffic are not encrypted.
security.protocol=sasl_sslSupportedSupportedRecommended SASL transport.
PLAINSupportedSupportedUse with TLS outside local tests.
SCRAM-SHA-256 / SCRAM-SHA-512SupportedSupportedRFC 5802/7677 non-PLUS exchange with server-signature verification.
GSSAPI, OAUTHBEARER, provider mechanismsNot supportedNot supportedUnsupported SASL mechanisms fail fast; there is no fallback client.

Compression is record-batch v2 compatible in both directions:

compression.type / fetched codecNative consumerNative producerNotes
noneSupportedSupported (default)No compression.
gzipSupportedSupportedStandard Kafka gzip record-batch compression.
snappySupportedSupportedStandard Kafka Snappy record-batch compression.
lz4SupportedSupportedStandard Java-compatible LZ4 frames with independent 64 KiB blocks.
zstdSupportedSupportedStandard Zstd frames at the library default level.

The decoder expands each compressed record batch through a pull-based reader into a buffer capped at 64 MiB. A malformed frame or expansion-limit breach returns a typed native-client error; it does not panic or allocate from an untrusted declared size without a bound.

Broker/version compatibility is explicit: the native path requires ApiVersions negotiation against Kafka 2.8+ and rejects broker.version.fallback or api.version.request=false. Classic consumer groups are supported with cooperative-sticky assignment by default. KIP-848 group.protocol=consumer, pattern subscriptions, Kafka transactions, and read_committed isolation are rejected with typed configuration errors.

Native group semantics ​

Native commits are broker offsets unless you choose CommitPolicy::External. CommitPolicy::Manual keeps the at-least-once rule: confirm your checkpoint or side effect first, then call commit() on the record or batch offset. CommitPolicy::AutoKafka is rejected because Kafka would own progress independently of your downstream durability.

Offset advancement is gap-blocking per partition. The native commit state tracks emitted, processed, and committed ranges, and it only sends the highest contiguous processed watermark to Kafka. If offsets 10 and 12 are confirmed but 11 is still outstanding, the broker commit does not move past 10. Drain and source close wait for outstanding offsets, flush the final due watermark, and complete only after the final broker commit response when Kafka commits are in use.

Classic consumer groups are implemented natively: FindCoordinator, JoinGroup/SyncGroup, Heartbeat, LeaveGroup, offset fetch/commit, generation/member tracking, cooperative-sticky assignment, eager range as the simple fallback, and static membership through group.instance.id. On revoke, native stops emitting revoked partitions, force-commits their processed watermark, and then releases ownership. On lost assignment, later commits fail as MqError::AssignmentLost rather than committing into partitions the group no longer owns.

Producing — idempotent retries with delivery confirmation ​

KafkaSink::plain produces ProducerRecords. Each record carries its own target topic; the sink enqueues bounded 256-record batches into the native producer owner and fails the stream if delivery fails. Production defaults to acks=all and enable.idempotence=true. The negotiated producer id/epoch and per-partition sequences let Kafka deduplicate retried batches. Set enable.idempotence=false to opt out explicitly; this permits bounded at-least-once retries that can duplicate after response loss.

rust
use datum::Sink;
use datum_mq::{KafkaProducerControl, ProducerRecord};

// Idempotent producer (enable.idempotence + acks=all) is the default from
// KafkaProducerSettings::new. The record carries its own target topic.
let sink: Sink<ProducerRecord, KafkaProducerControl> =
    KafkaSink::plain(KafkaProducerSettings::new("127.0.0.1:9092"));

let _ = sink;

The materialized KafkaProducerControl is how you confirm delivery at shutdown:

  • drain_and_shutdown() — wait for every accepted record to receive its delivery acknowledgement, then flush. Call this before dropping the sink so you don't lose in-flight records.
  • flush() — flush and drain outstanding deliveries.
  • metrics() — produced, in-flight, delivery failures, queue-full counters.
rust
// Requires a Kafka broker. Adapted from tests/kafka_integration.rs.
use datum::Source;
use datum_mq::{KafkaProducerSettings, KafkaSink, ProducerRecord};

let records = vec![ProducerRecord::new(
    "orders",
    Some(Vec::from(&b"created"[..]).into()),
)];

let control = Source::from_iterable(records)
    .run_with(KafkaSink::plain(
        KafkaProducerSettings::new("127.0.0.1:9092")
            .with("compression.type", "lz4"), // none, gzip, snappy, lz4, or zstd
    ))?;

control.drain_and_shutdown()?; // block until all deliveries are acknowledged

Rebalance behavior ​

Consumer-group rebalances preserve the public safety rule:

  • On revoke, the source force-flushes a synchronous offset commit for the revoked partitions before they move when Kafka commits are in use, so committed progress is not lost.
  • On a lost assignment (e.g. session timeout), the partitions are recorded and any later commit() for them returns MqError::AssignmentLost rather than committing into a partition the group no longer owns — the safe, correct failure.
  • Assigned/revoked/lost partitions and rebalance counts surface on KafkaMetricsSnapshot for correctness monitoring.

The native backend implements the classic group protocol directly and uses cooperative-sticky assignment for the benchmarked group path. KIP-848 remains outside the native boundary.

Configuration passthrough, TLS, and SASL ​

KafkaConfig is a raw property map. The native client uses Kafka-style keys for the ratified security subset:

rust
use datum_mq::{KafkaConfig, KafkaConsumerSettings};

// TLS, including optional mTLS certificate/key paths.
let tls = KafkaConfig::new("broker:9093").with_tls("/ca.pem", "/cert.pem", "/key.pem");

// SASL/SSL PLAIN.
let sasl = KafkaConfig::new("broker:9093").with_sasl_plain("user", "pass");

// SCRAM uses the raw Kafka client convention.
let scram = KafkaConfig::new("broker:9093")
    .with("security.protocol", "sasl_ssl")
    .with("ssl.ca.location", "/ca.pem")
    .with("sasl.mechanism", "SCRAM-SHA-512")
    .with("sasl.username", "user")
    .with("sasl.password", "pass");

// Or any raw property on the settings builder:
let settings = KafkaConsumerSettings::new("broker:9093", "group")
    .with("fetch.min.bytes", "1")
    .with("session.timeout.ms", "45000");

Native supports plaintext, ssl, sasl_plaintext, and sasl_ssl with PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512. sasl_plaintext is intended only for local tests because credentials and application traffic are not encrypted. TLS uses rustls with system roots plus optional ssl.ca.location/ssl.ca.pem, validates the advertised broker host through SNI, and accepts an mTLS certificate/key via the ssl.certificate.* and ssl.key.* properties. PLAIN should be used over TLS. SCRAM uses Kafka's non-PLUS mechanism and the RFC 5802 GS2 n flag because Kafka does not expose channel binding for these mechanisms. Broker-advertised session lifetimes trigger re-authentication on the existing connection.

Unsupported mechanisms fail fast. SASL GSSAPI, SASL OAUTHBEARER, provider plugins, delegation tokens, encrypted private keys, and non-rustls verification policies are outside the native support boundary.

Performance ​

datum-mq is benchmarked against Pekko Connectors Kafka 1.1.0 on a single-node KRaft broker (Kafka 4.2.0), 1,000,000 records × 256 B, 16 partitions. Full round-by-round method and raw numbers are in roadmap/benchmarks/mq.md.

KC-4 is the published-crate adapter row: fresh broker, same-session A/B, one warmup and three measured iterations, counters clean (consumed=unique=committed=end=1,000,000, duplicates=0, commit_ok=true, no_loss=true). CPU is whole-process milliseconds; RSS is peak resident set. The pre-v0.11.0 FFI-client rows are retained only as historical comparison numbers. They are not a selectable backend in v0.11.0.

ScenarioPekkoHistorical FFI-client row (removed)datum-mq nativeHonest read
consume_committable_1Mwall p50/p99 2,329.851/2,864.320 ms; CPU 2,400 ms; RSS peak 1,951,948 KiBwall p50/p99 2,104.547/2,208.612 ms; CPU 2,710 ms; RSS peak 386,496 KiBwall p50/p99 2,299.984/2,392.061 ms; CPU 1,720 ms; RSS peak 327,376 KiBNative beats Pekko on wall p50/p99 and CPU, and beats the historical FFI row on CPU/RSS, but loses to that removed row on wall p50/p99.
e2e_latency_1Mwall p50/p99 4,722.438/5,322.990 ms; latency p50/p99 4.773/9.151 ms; CPU 14,500 ms; RSS peak 1,792,440 KiBwall p50/p99 3,805.536/3,964.183 ms; latency p50/p99 137.438/363.036 ms; CPU 9,140 ms; RSS peak 233,428 KiBwall p50/p99 3,186.930/3,256.021 ms; latency p50/p99 1.788/6.777 ms; CPU 5,120 ms; RSS peak 39,228 KiBNative wins wall, CPU, RSS, and latency p50/p99 against both rows.
rebalance_disruption_1Mwall p50/p99 34,714.154/34,844.183 ms; CPU 3,390 ms; pause 1.642 ms; RSS peak 2,023,816 KiBwall p50/p99 4,473.221/4,607.881 ms; CPU 2,680 ms; pause 2.798 ms; RSS peak 372,256 KiBwall p50/p99 1,982.586/4,976.025 ms; CPU 1,610 ms; pause 1.455 ms; RSS peak 313,484 KiBNative is roughly 17.5x faster than Pekko at wall p50 and beats the historical FFI row on p50/CPU/RSS/pause, but native p99 is 368.144 ms above that removed row in this run.

Native plaintext produce's measured wall trade was accepted for its CPU, RSS, thread-count, and build-surface wins. Idempotence overhead measured within run-to-run noise, so enabling it on the native default does not change that accepted wall trade.

Deferred scope ​

  • Kafka transactions and consume-transform-produce EOS. The first stable guarantee is at-least-once; transactional exactly-once is a later work package.
  • SASL GSSAPI and OAUTHBEARER. PLAIN and SCRAM-SHA-256/512 are the supported SASL mechanisms.
  • KIP-848 consumer groups and pattern subscriptions. Classic groups with cooperative-sticky assignment are supported today.

See also ​

  • Context Propagation — SourceWithContext and offset handling.
  • CDC — pair a Postgres CDC source with a KafkaSink.
  • Agent — run a Kafka consumer as a supervised, drainable job.