partitionline

A pure-Rust Apache Kafka client and protocol implementation.

[dependencies]
# crates.io publish is one owner step away (docs/RELEASE.md → tag v0.1.0).
# Until then, pin a release-candidate tag (not floating main):
partitionline = { git = "https://github.com/mingley/partitionline", tag = "v0.1.0-rc.6" }
# After publish: partitionline = "0.1"

Status (WP-0.5): local civilization-check is green; crates.io publish waits on owner CARGO_REGISTRY_TOKEN + Actions capacity. Probe anytime: bash scripts/owner-status.sh (see docs/ADOPTION.md).

Produce

# async fn example() -> partitionline::Result<()> {
use partitionline::{ProduceRecord, Producer};

let producer = Producer::connect("127.0.0.1:9092").await?;
let md = producer
    .send(ProduceRecord::to("events").value(&b"hello"[..]))
    .await?;
println!("{}-{}@{}", md.topic, md.partition, md.offset);
producer.close().await?;
# Ok(())
# }

send waits for that record's offset. For many records, send_all queues then waits; try_send plus flush is the throughput path.

Fetch

# async fn example() -> partitionline::Result<()> {
use partitionline::Consumer;

let mut consumer = Consumer::connect("127.0.0.1:9092").await?;
consumer.assign("events", 0, 0).await?;
let recs = consumer.fetch().await?;
# let _ = recs;
# Ok(())
# }

fetch / group poll return ConsumerRecords (Java count / partitions / records / nextOffsets). Share poll returns ShareRecords. assign_topic assigns every partition. seek / pause / resume / wakeup match the Java consumer. Fetch talks to every partition leader; fenced partitions recover with OffsetForLeaderEpoch.

Groups

# async fn example() -> partitionline::Result<()> {
use partitionline::{ConsumerConfig, ConsumerGroup};

let mut group = ConsumerGroup::join_topics(
    ConsumerConfig::bootstrap(["127.0.0.1:9092"]),
    "workers",
    ["orders", "payments"],
)
.await?;
let recs = group.poll().await?;
group.commit_with_metadata(recs.next_offsets()).await?;
group.leave().await?;
# Ok(())
# }

Classic range, sticky, cooperative-sticky (KIP-429), KIP-848 (join_consumer), and KIP-932 share groups (ShareGroup::join / poll / accept / release / reject). auto_offset_reset is used when the group has no committed offset (Earliest by default, unlike Java's latest).

Transactions

# async fn example() -> partitionline::Result<()> {
use partitionline::{ProduceRecord, Producer, ProducerConfig};

let producer = Producer::new(
    ProducerConfig::bootstrap(["127.0.0.1:9092"]).transactional_id("orders-txn"),
)
.await?;
producer.init_transactions().await?;
producer.begin_transaction().await?;
producer
    .send(ProduceRecord::to("events").value(&b"hello"[..]))
    .await?;
producer.commit_transaction().await?;
producer.close().await?;
# Ok(())
# }

transactional.id implies idempotence. send_offsets_to_transaction takes TopicPartition. send_offsets_for_group uses the group's ConsumerGroupMetadata (see examples/eos.rs).

Admin

# async fn example() -> partitionline::Result<()> {
use partitionline::{Admin, NewTopic};

let mut admin = Admin::connect("127.0.0.1:9092").await?;
admin
    .create_topics(&[NewTopic::new("events", 1, 1)], 30_000, false)
    .await?;
let topics = admin.list_topics_with(false).await?;
# let _ = topics;
admin.close().await?;
# Ok(())
# }

CreateTopics, DeleteTopics, DescribeConfigs, IncrementalAlterConfigs, ACLs, groups, transactions, log dirs, quotas, and the other Kafka 3.x / 4.x admin APIs are on Admin. Method-level rustdoc names the matching Java Admin / *Options calls.

Configure

Builders, not a property bag of strings:

use std::time::Duration;
use partitionline::{Acks, Compression, IsolationLevel, ProducerConfig, Sasl};

let _cfg = ProducerConfig::bootstrap(["127.0.0.1:9092"])
    .acks(Acks::All)
    .linger(Duration::from_millis(5))
    .compression(Compression::Lz4)
    .sasl(Sasl::scram_sha256("alice", "secret"));
let _iso = IsolationLevel::ReadCommitted;

TLS is TlsConfig (rustls, no OpenSSL). SASL PLAIN, SCRAM-SHA-256/512, and OAUTHBEARER (including OIDC) are supported. Compression is gzip, snappy, and lz4. Produce partitioning is murmur2 / round-robin; override it with ProducerConfig::partitioner.

Defaults that differ from Java:

• auto.offset.reset is Earliest (Java latest)
• allow.auto.create.topics is false (Java consumer true)
• delivery.timeout.ms is 30s (Java 120s)
• max.block.ms is 30s (Java 60s)

buffer.memory (32 MiB) and max.request.size (1 MiB) match Java. retry.backoff.ms / reconnect.backoff.ms / connections.max.idle.ms / metadata.max.age.ms / transaction.timeout.ms match Java.

The crate rustdoc is the full Java-shaped API catalog (protocol helpers, version ranges, and option names). Operator guide: docs/guide.md. Migrate from rust-rdkafka: docs/migrate-from-rdkafka.md. Adoption / pilot checklist: docs/ADOPTION.md. Capability list vs librdkafka: docs/gaps.md. Security: docs/security.md. Release policy: docs/RELEASE.md. Roadmap: docs/CIVILIZATION.md.

Not a drop-in for rd_kafka_*. Still missing vs librdkafka: zstd and Kerberos (C libraries), Schema Registry.

Features vs C stack

Examples

Broker on 127.0.0.1:9092 (Docker apache/kafka:3.9.1 is enough):

cargo run --release --example roundtrip

Also: produce, consume, group, txn, admin, sasl, oauth, tls, eos, offsets, share, wakeup, pause, metrics, cooperative, intercept, consume_intercept.

Locked produce vs librdkafka 2.15.0 C (linger 5ms, 8e6×100B). Do not publish rec/s unless broker high watermark equals records sent:

COUNT=8000000 WARMUP_SECS=0 PAYLOAD_BYTES=100 ACKS=1 LINGER_MS=5 KAFKA_TOPIC=plbench \
  cargo run --release --example bench_produce

Produce C bar (Lab A), fetch writeup (this-VM, unsigned), and latency writeup (this-VM, unsigned): docs/benchmark.md. Suite HOLD: docs/STATUS.md.

Numbers

Locked produce vs librdkafka 2.15.0 C is Lab A (broker high watermark equals records sent). Latency writeup is this-VM 2026-08-28, unsigned.

Fetch writeup 2026-08-28 this-VM (Apache Kafka 3.9.1 KRaft; consumed equals sent, HW 8e6). Comparison is rust-rdkafka 0.39.0 BaseConsumer::poll (bundled librdkafka 2.12.1), not Lab A and not C 2.15.0. Unsigned until Kernel Integrity signs. Not a Suite HOLD lift.

Latency writeup 2026-08-28 this-VM (Apache Kafka 3.9.1 KRaft). Sequential produce-ack, linger 0, 10k × 100 B, 1 partition, HW 11,000 (1k warmup + 10k timed). Comparison is rust-rdkafka 0.39.0 FutureProducer (bundled librdkafka 2.12.1), not Lab A and not C 2.15.0. Unsigned until Kernel Integrity signs. Not a Suite HOLD lift. Not a "faster than librdkafka" claim: this-VM p50/p99 medians are not a win.

COUNT=10000 WARMUP=1000 PAYLOAD_BYTES=100 ACKS=1 LINGER_MS=0 \
  MODE=both KAFKA_TOPIC=pllat \
  cargo run --release --example bench_latency

License

MIT OR Apache-2.0