Magnetar

A blazing-fast, async, sans-io Apache Pulsar client for Rust.

Status: stable (1.7.1). Full Apache Pulsar Java-client parity with a sans-io protocol core and two interchangeable engines — a production tokio engine (usable end-to-end, with supervised reconnect + transparent producer/consumer rebuild) and a deterministic-simulation moonpool engine (client/producer/consumer). The public API follows Semantic Versioning. PIP-460 scalable topics remain experimental and are excluded from the 1.0 stability promise.

---

What is magnetar?

Magnetar is a from-scratch Apache Pulsar client driver written in Rust. It mirrors the surface area of the Apache Pulsar Java client and adds two properties that the Java client cannot reach:

1. Sans-io core. The protocol state machine (magnetar-proto) is a pure, quinn-proto-style state machine — handle_bytes in, poll_transmit out, poll_event for semantic events, poll_timeout for timers. Zero I/O dependencies. No tokio. No async. No sockets. It is feed-only.
2. Multiple swappable engines. The same sans-io state machine is driven by a production tokio engine (magnetar-runtime-tokio) and by a deterministic simulation engine (magnetar-runtime-moonpool) for chaos testing of reconnects, partitions, and TLS handshake reorderings under reproducible seeds.

The architecture explicitly bans channels (mpsc, broadcast, watch, oneshot, crossbeam-channel, flume, async-channel, …). The wake-up mechanism is Arc<parking_lot::Mutex<State>> plus tokio::sync::Notify plus core::task::Waker slabs inside the state machine. See ARCHITECTURE.md for the full rationale.

Magnetar is independent of the existing pulsar-rs crate — it shares neither code nor dependencies. The goal is feature-complete parity with the Apache Pulsar Java client.

---

Features at a glance

• Protocol coverage: producer, consumer, reader, partitioned producer, partitioned consumer, multi-topics consumer, pattern (regex) consumer, table view, transactions.
• PIPs implemented or partially wired: PIP-4 (end-to-end encryption), PIP-30 / PIP-292 (in-band AUTH_CHALLENGE refresh), PIP-31 (transactions), PIP-37 (chunking — bounded consumer-side reassembly, cap 10 / 60s expiry, matching Java — plus redelivery backoff), PIP-54 (partial-batch ACK), PIP-74 (auto-scaled receiver queue — pluggable ReceiverQueuePolicy, Fixed default / opt-in Auto), PIP-87 (AutoConsumeSchema broker lookup), PIP-90 (broker-entry metadata), PIP-121 (cluster failover — ServiceUrlProvider + ControlledClusterFailover
  • AutoClusterFailover), PIP-145 (regex topic discovery), PIP-188 (TOPIC_MIGRATED with supervised reconnect), PIP-313 (force unsubscribe). See Supported PIPs.
• Resilience: supervised reconnect with Connection::reset (Stage 2) + producer / consumer rebuild via rebuild_producers / rebuild_consumers (Stage 3) + memory_limit runtime enforcement for both Java policies + global publish-bytes accounting via AtomicU64 CAS in Producer::send with release on Drop. Single and chunked publishes replay transparently; non-replayable batched sends fail deterministically, established durable consumers defer to the broker cursor, and non-durable consumers reuse only their original start position on reattach (ADR-0096, ADR-0099).
• Observability: cumulative counters + hdrhistogram p50/p99/max latency
  • rolling-window msgs/sec + bytes/sec rates (record_rate_window). OpenTelemetry context propagation (traceparent/tracestate via message properties, feature = "opentelemetry", ADR-0053).
• Transports: TCP, TLS 1.3 (rustls-only — no native-tls, no openssl), binary proxy (proxy_to_broker_url), pluggable DNS (DnsResolver trait + TokioDnsResolver default routed through Transport::connect).
• TLS knobs: tls_trust_certs_file_path, tls_allow_insecure_connection (blanket override), tls_hostname_verification_enable(false) paired with a PEM trust store (chain-on / hostname-off via custom rustls verifier).
• Schemas: bytes, string, JSON, Avro, Protobuf, Protobuf-native, KeyValue, Auto-consume, Auto-produce-bytes, plus the full primitive family — Int8, Int16, Int32, Int64, Float, Double, Bool, Date, Time, Timestamp, LocalDate, LocalTime, Instant, LocalDateTime.
• Compression: LZ4, ZSTD, Snappy, ZLIB.
• Auth providers: token, mTLS (the two stock providers in magnetar-proto::auth), OAuth2 ClientCredentialsFlow (working — fetches
  • caches + auto-refreshes JWTs against a standard OIDC token endpoint), SASL PLAIN (RFC 4616, working), SASL Kerberos / GSSAPI via libgssapi under the auth-sasl-kerberos feature (working — multi-round AUTH_CHALLENGE initiate loop), Athenz with a pre-fetched role token (AthenzProvider::with_role_token, working), and the opt-in Athenz ZTS round-trip (auth-athenz-zts, working via AthenzProvider::with_default_signer or a custom zts::JwtSigner / zts::ZtsClient).
• Trackers: ack grouping, unacked-message tracker (ack timeout + redelivery), negative-ack tracker with MultiplierRedeliveryBackoff (PIP-37), batch-index ACK set (PIP-54) with conservative post-reset reconstruction.
• Interceptors: ProducerInterceptor + ConsumerInterceptor SPIs.
• Admin REST client: a reqwest-backed admin client lives in magnetar-admin.
• CLI: the magnetarctl binary in the magnetarctl crate provides data-plane produce / consume commands and the admin lookup, policy, and stats surface.

---

Installation

Magnetar's current release is 1.7.1. The façade is published to crates.io under the package name magnetar-driver (the magnetar name is held by an unrelated crate); its library / import name stays magnetar, so use magnetar::... is unchanged.

Depend on it directly from crates.io:

[dependencies]
# crates.io package is `magnetar-driver`; the import path stays `magnetar`.
magnetar-driver = "1.7.1"

Or pin the tagged release via Git:

[dependencies]
magnetar-driver = { git = "https://github.com/CleverCloud/magnetar", tag = "v1.7.1" }

The default feature set enables the tokio engine. The feature flags catalog:

Flag	Default	Effect
tokio	yes	Pulls in magnetar-runtime-tokio plus tokio/futures-util. The public PulsarClient lives behind this flag.
moonpool	no	Pulls in magnetar-runtime-moonpool for deterministic-simulation testing.
admin	no	Re-exports magnetar-admin under magnetar::admin.
auth-oauth2	no	Pulls in magnetar-auth-oauth2 (OAuth2 ClientCredentialsFlow provider).
auth-sasl	no	Pulls in magnetar-auth-sasl (SASL PLAIN + the sans-io Kerberos surface).
auth-sasl-kerberos	no	Implies auth-sasl and turns on magnetar-auth-sasl/kerberos, which binds libgssapi. Build host needs the MIT KRB5 / Heimdal headers (krb5-devel / libkrb5-dev) and libclang (clang-libs / libclang-dev) — libgssapi-sys runs bindgen at build time. See ADR-0029.
auth-athenz	no	Pulls in magnetar-auth-athenz.
auth-athenz-zts	no	Implies auth-athenz and turns on the reqwest-backed ZTS exchange plus in-tree JWT signer support.
encryption	no	Pulls in magnetar-messagecrypto plus the PIP-4 bridge type.

| experimental-v5-client | no | Enables the PIP-466 V5 wrapper surface (magnetar::v5) over the v4 wire commands. | | scalable-topics | no | Enables the experimental PIP-460 scalable-topic surface (topic://, layout sessions, StreamConsumer, topic-info). Negotiated per connection; inert against a Pulsar 4.x broker. | | crypto-aws-lc-rs | yes | rustls crypto provider: aws-lc-rs; brings post-quantum hybrid KEX (X25519MLKEM768). See TLS crypto provider. | | crypto-ring | no | rustls crypto provider: ring. | | crypto-openssl | no | rustls crypto provider: rustls-openssl (wraps system OpenSSL via deny.toml carve-out). | | crypto-fips | no | rustls crypto provider: aws-lc-rs FIPS-validated module (requires cmake + C toolchain). |

The workspace ships twelve crates:

Crate	Role
magnetar	Public façade — re-exports + builder + typed schemas wiring. crates.io package magnetar-driver; library / import name magnetar.
magnetar-proto	Sans-io protocol crate. The heart of the project.
magnetar-runtime-tokio	Production tokio engine with tokio-rustls TLS.
magnetar-runtime-moonpool	Deterministic-simulation engine (rustls-over-bytepipe TLS, no native TLS).
magnetar-differential	Tokio ↔ moonpool EventStream equivalence harness (cross-engine tests).
magnetar-admin	REST admin client (reqwest + rustls-tls).
magnetarctl	magnetarctl binary — admin lookups today, produce / consume / inspect coming.
magnetar-fakes	In-process broker fake (dev-dep). Mirrors Java's MockBrokerService.
magnetar-auth-oauth2	OAuth2 ClientCredentialsFlow auth provider.
magnetar-auth-sasl	SASL auth provider.
magnetar-auth-athenz	Athenz auth provider.
magnetar-messagecrypto	PIP-4 end-to-end encryption (AES-GCM via aws-lc-rs).

xtask is a workspace member but is not published — it hosts build helpers (protoc codegen, e2e driver, dependency audits).

---

TLS crypto provider

The rustls crypto backend is selected at compile time via four mutually-pluggable Cargo features on the magnetar façade. The wire protocol — TLS 1.3 (default) / TLS 1.2 — is identical across every provider; what differs is the audited / FIPS-validated / post-quantum posture of the underlying primitives.

Feature	Backend	Post-quantum KEX	FIPS validated	Pure Rust	Default
crypto-aws-lc-rs	aws-lc-rs	yes (X25519MLKEM768)	no	no (C)	✓
crypto-ring	ring	no	no	no (C)
crypto-openssl	rustls-openssl	yes	depends on OpenSSL build	no
crypto-fips	aws-lc-fips-sys	(FIPS-approved only)	yes	no (C)

# Pick a single provider (mutually exclusive at build time).
# Façade crates.io package is `magnetar-driver`; import path stays `magnetar`.
cargo build -p magnetar-driver --no-default-features --features tokio,crypto-aws-lc-rs
cargo build -p magnetar-driver --no-default-features --features tokio,crypto-ring
cargo build -p magnetar-driver --no-default-features --features tokio,crypto-openssl   # needs system OpenSSL
cargo build -p magnetar-driver --no-default-features --features tokio,crypto-fips      # needs cmake + C toolchain

# The `magnetarctl` binary (magnetarctl crate) mirrors the same cascade — the
# admin REST client (reqwest + rustls) and the data-plane runtime both
# bind to the selected provider. `cargo build -p magnetarctl` alone
# defaults to `crypto-aws-lc-rs`.
cargo build -p magnetarctl --no-default-features --features crypto-ring

Under cargo build --workspace --all-features the compile-time cfg cascade resolves to aws-lc-rs (highest priority). Single-provider builds go through cargo run -p xtask -- check-crypto-matrix. A single compile_error! fires if no crypto-* feature is enabled.

The crypto-aws-lc-rs default picks up rustls 0.23's built-in prefer-post-quantum feature, so the wire client negotiates the X25519MLKEM768 hybrid key exchange with brokers that support it.

openssl / openssl-sys are admitted only as transitive deps of rustls-openssl; the rest of ADR-0005 (no native-tls, rustls everywhere) stays in force. See ADR-0035 for the binding decision.

---

Build metadata (magnetarctl --version)

The magnetarctl binary exposes a sozu / systemd-style identification banner:

$ magnetarctl --version
magnetarctl 1.7.1 (a1b2c3d4e5f6)
built 2026-05-26T14:32:11Z · profile=release · rustc=rustc 1.91.0 (…) · target=x86_64-unknown-linux-gnu
features: +default
pulsar wire protocol: v21
os: linux · report bugs at https://github.com/CleverCloud/magnetar

• -V prints a single-line, never-colorized form: magnetarctl 1.7.1 (sha).
• --version prints the multi-line form above, colorized on a TTY. NO_COLOR=1 or piping suppresses ANSI (https://no-color.org).
• SOURCE_DATE_EPOCH=<unix-seconds> pins the build timestamp for reproducible builds.

Full reference: docs/cli.md.

---

Quickstart

The high-level PulsarClient builder is the public entry point. It wires the tokio engine to the sans-io state machine and gives you producer / consumer / reader / table-view / partitioned / multi-topics / pattern builders.

Producer + consumer round trip

use magnetar::{OutgoingMessage, PulsarClient};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let producer = client
    .producer("persistent://public/default/orders")
    .name("orders-writer")
    .compression(magnetar_proto::types::CompressionKind::Zstd)
    .batching(/* max_messages */ 256, /* max_bytes */ 128 * 1024)
    .create()
    .await?;

producer
    .send(OutgoingMessage::with_payload(b"hello, pulsar".as_slice()).into())
    .await?;

let consumer = client
    .consumer("persistent://public/default/orders")
    .subscription("worker")
    .subscription_type(magnetar_proto::pb::command_subscribe::SubType::Shared)
    .subscribe()
    .await?;

let msg = consumer.receive().await?;
println!("payload: {:?}", msg.payload);
consumer.ack(msg.message_id).await?;
# Ok(()) }

Typed schemas (TypedProducer + TypedConsumer)

use std::sync::Arc;
use magnetar::{PulsarClient, TypedProducerBuilder, TypedConsumerBuilder};
use magnetar_proto::schema::StringSchema;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let schema = Arc::new(StringSchema::new());

let producer = client
    .typed_producer("persistent://public/default/notes", schema.clone())
    .create()
    .await?;

producer.new_message().value("a note".to_string()).send().await?;

let consumer = client
    .typed_consumer("persistent://public/default/notes", schema)
    .subscription("transcriber")
    .subscribe()
    .await?;

let msg = consumer.receive().await?;
println!("decoded value: {}", msg.value);
consumer.ack(msg.id).await?;
# Ok(()) }

Reader (non-durable, exclusive)

use magnetar::PulsarClient;
use magnetar_proto::MessageId;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let reader = client
    .reader("persistent://public/default/events")
    .start_message_id(MessageId::EARLIEST)
    .create()
    .await?;

while let Ok(msg) = reader.receive().await {
    println!("entry {:?}", msg.message_id);
}
# Ok(()) }

Partitioned producer + consumer

use magnetar::{PulsarClient, MessageRoutingMode};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let p = client
    .partitioned_producer("persistent://public/default/events")
    .routing_mode(MessageRoutingMode::RoundRobin)
    .batching(/* max_messages */ 128, /* max_bytes */ 64 * 1024)
    .create()
    .await?;

p.new_message().key("user-42").value(b"event".as_slice()).send().await?;

let c = client
    .partitioned_consumer("persistent://public/default/events")
    .subscription("workers")
    .dead_letter_policy(3, None)
    .subscribe()
    .await?;

let dlq = client
    .producer("persistent://public/default/events-DLQ")
    .create()
    .await?;

// After prior failed deliveries have populated the children's dead-letter buffers:
let republished = c.republish_dead_letters(&dlq).await?;
println!("republished {republished} dead letters across all partitions");
# Ok(()) }

Pattern (regex) consumer — PIP-145

use magnetar::PulsarClient;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let pc = client
    .pattern_consumer()
    .namespace("public/default")
    .pattern("orders-.*")
    .subscription("workers")
    .subscribe(&client)
    .await?;

println!("matched topics: {:?}", pc.topics());

let msg = pc.receive().await?;
pc.ack(msg.topic(), msg.message_id).await?;
# Ok(()) }

Table view (latest-value-per-key snapshot from a compacted topic)

use magnetar::PulsarClient;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let view = client
    .table_view("persistent://public/default/config")
    .subscription("cfg-watcher")
    .create()
    .await?;

view.for_each(|key, value| println!("{key} = {value:?}"));
let last = view.get("api.threshold");
# Ok(()) }

Transactions — PIP-31

use std::time::Duration;
use magnetar::{PulsarClient, OutgoingMessage};

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

// open the transaction-coordinator-backed transaction
let runtime_client = /* obtain magnetar_runtime_tokio::Client */
#   unreachable!();
let txn = runtime_client.new_txn(Duration::from_mins(1)).await?;

let producer = client
    .producer("persistent://public/default/orders")
    .create()
    .await?;

producer
    .send(OutgoingMessage::with_payload(b"line-item".as_slice()).txn(txn.id()).into())
    .await?;

txn.commit().await?;
# Ok(()) }

Interceptors (ProducerInterceptor + ConsumerInterceptor)

use std::sync::Arc;
use magnetar::{
    ConsumerInterceptor, IncomingMessage, OutgoingMessage, ProducerInterceptor, PulsarClient,
    send_with_interceptors,
};

#[derive(Debug)]
struct StampSender;
impl ProducerInterceptor for StampSender {
    fn before_send(&self, msg: &mut OutgoingMessage) {
        msg.properties.push(("client".to_owned(), "magnetar".to_owned()));
    }
}

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = PulsarClient::builder()
    .service_url("pulsar://localhost:6650")
    .build()
    .await?;

let producer = client
    .producer("persistent://public/default/orders")
    .create()
    .await?;

let chain: Vec<Arc<dyn ProducerInterceptor>> = vec![Arc::new(StampSender)];
let id = send_with_interceptors(
    &producer,
    OutgoingMessage::with_payload(b"hi".as_slice()),
    &chain,
)
.await?;
println!("acked at {id:?}");
# Ok(()) }

---

Java client parity matrix

A check (✅) is a working public-API surface backed by code in the workspace. A flag (🟡) means partial — a working subset; check ARCHITECTURE.md for the open gaps. A cross (❌) is a known-missing feature.

Producer

Feature	Java	Magnetar	Notes
send(...) / sendAsync(...)	✅	✅	Producer::send returns a SendFut.
Producer name	✅	✅	ProducerBuilder::name.
Compression (LZ4, ZSTD, Snappy, ZLIB, NONE)	✅	✅	ProducerBuilder::compression. The production tokio engine supports all listed codecs; the Moonpool engine currently accepts only NONE and rejects non-None compression until its provider-native codec path lands.
Batching (BatchMessageContainerImpl)	✅	✅	ProducerBuilder::batching (max-msgs + max-bytes). One ranged wire frame has no safe per-message replay representation, so a reconnect before its receipt resolves every affected SendFut with a bounded send error instead of replaying or orphaning it (ADR-0096).
batchingMaxPublishDelay flush timer	✅	✅	ProducerBuilder::batching_max_publish_delay.
Chunking (PIP-37)	✅	✅	ProducerBuilder::chunking. Chunks-never-batched enforced. Consumer-side reassembly is bounded (ConsumerBuilder::max_pending_chunked_message default 10, expire_time_of_incomplete_chunked_message default 60s, auto_ack_oldest_chunked_message_on_queue_full default false), matching the Java client; the oldest incomplete message is evicted on cap breach and stale buffers are swept on the timeout tick, so a broker streaming distinct-UUID first chunks cannot grow the reassembly map without bound. Accepted incomplete chunks also use Java-compatible per-chunk flow replenishment, repaying each consumed broker permit before logical-message reassembly completes.
initialSequenceId	✅	✅	ProducerBuilder::initial_sequence_id.
sendTimeout	✅	✅	ProducerBuilder::send_timeout. Defaults to 30 s (Java parity — sendTimeoutMs = 30000, ADR-0072); a send whose receipt is lost/corrupted in flight fails with a code=-1, "send timeout" error instead of hanging. ProducerBuilder::disable_send_timeout restores the unbounded behavior.
accessMode (Shared/Exclusive/WaitForExclusive/Fencing)	✅	✅	ProducerBuilder::access_mode. PIP-68.
accessMode getter	✅	✅	Producer::access_mode.
getProducerName	✅	✅	Producer::name.
getTopic	✅	✅	Producer::topic.
isConnected / isClosed	✅	✅	Producer::is_connected / is_closed.
getLastSequenceId	✅	✅	Producer::last_sequence_id.
getLastSequenceIdPublished	✅	✅	Producer::last_sequence_id_published.
getLastDisconnectedTimestamp	✅	✅	Producer::last_disconnected_timestamp.
flush()	✅	✅	Producer::flush.
close()	✅	✅	Producer::close. Last-clone Drop additionally fires a best-effort fire-and-forget close (ADR-0057) — beyond Java parity, where an abandoned producer leaks broker-side until disconnect.
getStats	✅	✅	Producer::stats — counters + send_latency_{p50,p99,max}_ms via hdrhistogram + rolling per-second msgs_per_sec / bytes_per_sec windows (producer_record_rate_window).
getCompressionType getter	✅	✅	Producer::compression.
Per-message key / orderingKey	✅	✅	OutgoingMessage::key / ordering_key.
Per-message eventTime	✅	✅	OutgoingMessage::event_time_ms.
deliverAt / deliverAfter	✅	✅	OutgoingMessage::deliver_at_ms / deliver_after_ms.
replicationClusters + disableReplication	✅	✅	OutgoingMessage::replication_clusters / disable_replication.
newMessage(Transaction) (PIP-31)	✅	✅	OutgoingMessage::txn(txn_id).
Properties (per-message key/value)	✅	✅	OutgoingMessage::property.
TypedMessageBuilder	✅	✅	MessageBuilder via ProducerExt::new_message.
ProducerInterceptor SPI	✅	✅	magnetar::ProducerInterceptor + send_with_interceptors.
OpenTelemetry context propagation	❌	✅	feature = "opentelemetry" — auto-inject traceparent/tracestate at send boundary (ADR-0053).
pendingQueueSize getter	✅	✅	Producer::pending_count (batch_len + batch_bytes are bonus).

Consumer

Feature	Java	Magnetar	Notes
subscribe(...) (Exclusive / Shared / Failover / Key_Shared)	✅	✅	ConsumerBuilder::subscription_type.
receive / receiveAsync / receive(timeout)	✅	✅	Consumer::receive + receive_with_timeout.
batchReceive / batchReceiveAsync	✅	✅	Consumer::receive_batch_with_bytes_cap (cap on count + bytes).
messageListener (push delivery)	✅	✅	ConsumerBuilder::message_listener(...) + subscribe_with_listener(). The typed twin delivers a decoded TypedMessage<S> directly. The poller drives receive() and invokes the callback sequentially, in order; the callback acks explicitly. Explicit or terminal remote close ends receive(). Dropping MessageListenerHandle aborts the poller and drops its owned consumer clone; only a final clone stages a best-effort close. Pull and push are mutually exclusive — subscribe_with_listener() moves the consumer into the poller, returning the listener handle. Full parity: single-topic + typed and the wrapper consumers — MultiTopicsConsumerBuilder / PartitionedConsumerBuilder / PatternConsumerBuilder each take message_listener(...) + subscribe_with_listener(), with a Fn(&str, &IncomingMessage) callback. Later-discovered pattern / partition children inherit the listener through the membership-change race. ADR-0064.
consumerEventListener (Failover becameActive/becameInactive)	✅	✅	ConsumerBuilder::consumer_event_listener(...) + subscribe_with_event_listener(). ConsumerEventListener = Arc<dyn Fn(ConsumerEvent) + Send + Sync> fires ConsumerEvent::BecameActive / BecameInactive once per broker CommandActiveConsumerChange, sequentially, from a detached poller task driving Consumer::next_active_change() — mirrors messageListener's poller shape exactly, but independent of message delivery (a consumer can hold both). Consumer::is_active() exposes the last-reported state synchronously (None until the first transition). Dropping ConsumerEventListenerHandle aborts the poller; close() awaits a clean stop. ADR-0081.
acknowledge (individual)	✅	✅	Consumer::ack.
acknowledgeCumulative	✅	✅	Consumer::ack_cumulative.
acknowledge(messages) (batch ack)	✅	✅	Consumer::ack_batch.
acknowledge(MessageId, Map<String,String>)	✅	✅	Consumer::ack_with_properties.
acknowledge(MessageId, Transaction) (PIP-31)	✅	✅	Consumer::ack_with_txn.
acknowledgeAsync(messages, Transaction)	✅	✅	Consumer::ack_batch_with_txn.
acknowledgeCumulative(MessageId, Map)	✅	✅	Consumer::ack_cumulative_with_properties.
acknowledgeCumulative(MessageId, Transaction)	✅	✅	Consumer::ack_cumulative_with_txn.
Batch-index ACK (PIP-54)	✅	✅	ack_set bitset stamped on individual acks. If reset has cleared the local tracker, a fresh all-unacked set is reconstructed and only the requested index is cleared; absence never becomes a full-entry ack (ADR-0096). The bitset the broker delivers on CommandMessage is read too: a re-dispatched batched entry surfaces only the positions it still lists as unacked, charges a permit for none of the others, and seeds / AND-accumulates the tracker entry from it (ADR-0105).
acknowledgmentGroupTime (grouping window)	✅	✅	ConsumerBuilder::ack_group_time + ack_grouped / ack_grouped_cumulative.
negativeAcknowledge	✅	✅	Consumer::negative_ack.
negativeAcknowledge(messages)	✅	✅	Consumer::negative_ack_batch.
negativeAcknowledge(MessageId, delay)	✅	✅	Consumer::negative_ack_with_delay.
MultiplierRedeliveryBackoff (PIP-37)	✅	✅	magnetar_proto::trackers::MultiplierRedeliveryBackoff.
reconsumeLater (retry-letter topic)	✅	✅	Consumer::reconsume_later + _with_properties.
ackTimeout (unacked tracker)	✅	✅	ConsumerBuilder::ack_timeout. Client-side redelivery timeout for messages the broker delivered but the app never acked — not to be confused with ack_response_timeout below (the wire round-trip deadline for a CommandAck the client already sent).
ackTimeoutRedeliveryBackoff (PIP-37)	✅	✅	ConsumerBuilder::ack_timeout_backoff.
ack_response_timeout (no Java equivalent)	❌	✅	ClientBuilder::ack_response_timeout / ClientBuilder::disable_ack_response_timeout (connection-wide, mirrors sendTimeout's rationale, ADR-0072). A CommandAck whose CommandAckResponse never arrives fails with code=-1, "ack timeout" instead of hanging the caller's ack().await forever. Defaults to 30 s; disabling restores the unbounded behavior. A same-broker CloseConsumer (bundle reassignment, #307) additionally fails every ack pending against the torn-down consumer id immediately, ahead of this deadline. (#346)
negativeAckRedeliveryDelay	✅	✅	ConsumerBuilder::negative_ack_redelivery_delay.
seek(MessageId)	✅	✅	Consumer::seek.
seek(timestamp)	✅	✅	Consumer::seek_timestamp.
seekAsync(Function<String, Object>) (per-partition)	✅	✅	PartitionedConsumer::seek_per_partition / MultiTopicsConsumer::seek_per_partition — callback returns SeekTarget::MessageId or SeekTarget::PublishTimeMs per topic.
seekToEarliest / seekToLatest	✅	✅	Consumer::seek_to_earliest / seek_to_latest.
pause() / resume() / isPaused()	✅	✅	Consumer::pause / resume / is_paused.
receiverQueueSize / autoScaledReceiverQueueSizeEnabled (PIP-74)	✅	✅	Pluggable ReceiverQueuePolicy (ADR-0071, issue #301). ConsumerBuilder::receiver_queue_size(n) pins a Fixed(n) queue (the default, byte-identical to before); ConsumerBuilder::receiver_queue_policy(Arc::new(Auto::new(min, max_bytes))) opts into the PIP-74 auto-scaled queue — grows by bounded doubling under starvation, shrinks under a buffered-bytes OOM guard. The adjust tick is pure + rides the sans-io clock so both engines stay bit-reproducible. Threaded through partitioned / multi-topics / pattern consumers. Current target readable via Consumer::current_receiver_queue_size.
hasReachedEndOfTopic	✅	✅	Consumer::has_reached_end_of_topic.
redeliverUnacknowledgedMessages	✅	✅	Consumer::redeliver_unacked.
getLastMessageId	✅	✅	Consumer::last_message_id.
getStats (counters)	✅	✅	Consumer::stats. Includes total_chunked_msgs_received and pending_batch_acks (magnetar-specific, no Java counterpart). MultiTopicsConsumer::aggregate_stats / PartitionedConsumer::aggregate_stats / PartitionedProducer::aggregate_stats (magnetar-specific — Java's PartitionedConsumerImpl exposes no aggregate stats getter) fold every child's snapshot via ConsumerStats::fold / ProducerStats::fold: totals and pending_batch_acks sum.
Stats: rolling windows (msgs/sec, bytes/sec)	✅	✅	ConsumerStats::msgs_per_sec / bytes_per_sec + ProducerStats same. Set ClientBuilder::stats_interval(dur) and the client samples every producer and consumer on that cadence — including the per-partition and per-topic children behind the wrapper types, so aggregate_stats() folds real rates (ADR-0089). The knob defaults to Some(60 s), matching Java. ClientBuilder::stats_interval(Duration::ZERO) disables the sweep and leaves sampling caller-driven: call Producer::record_rate_window(now) / Consumer::record_rate_window(now) (or Connection::{producer,consumer}_record_rate_window(handle, now)) on your own cadence — the first call records the baseline, subsequent calls compute per-second rates from the delta, and the fields stay 0.0 until you do. Pick one cadence: a manual call while the sweep is running re-seeds the window. aggregate_stats() folds the rates as an f64 sum across children (fan-in throughput).
Stats: latency hdrhistogram (p50/p99/max)	✅	✅	Consumer::stats exposes receive_latency_{p50,p99,max}_ms; Producer::stats exposes send_latency_{p50,p99,max}_ms. aggregate_stats() takes the exact max across children and recomputes p50/p99 from a real hdrhistogram::Histogram::add merge of every child's {receive,send}_latency_histogram() — not a sum/max of the children's own percentile fields, which is not statistically sound (#347).
statsInterval (client-driven rate sampling)	✅	✅	ClientBuilder::stats_interval(dur); Duration::ZERO disables, spelling Java's statsIntervalSeconds = 0. Java self-ticks each ProducerStatsRecorderImpl / ConsumerStatsRecorderImpl on the client-wide HashedWheelTimer; magnetar arms the same obligation as a deadline on the sans-io poll_timeout / handle_timeout loop, so it reaches every slot on the connection and the wrappers need no fan-out either (ADR-0089). Default Some(60 s), matching Java's statsIntervalSeconds. A producer or consumer created mid-window reports 0.0 for its first interval, exactly as Java's does.
subscriptionProperties	✅	✅	ConsumerBuilder::subscription_property.
replicateSubscriptionState	✅	✅	ConsumerBuilder::replicate_subscription_state.
priorityLevel	✅	✅	ConsumerBuilder::priority_level.
keySharedPolicy (sticky / auto-split / hash)	✅	✅	ConsumerBuilder::key_shared_policy. PIP-34/119/282/379.
startMessageId	✅	✅	ConsumerBuilder::start_message_id. The explicit value applies to initial attachment; an established durable reattach omits it and uses the broker cursor, while a non-durable reattach reuses only the caller's original position and never a locally submitted ack watermark (ADR-0096, ADR-0099).
startMessageRollbackDuration	✅	✅	ConsumerBuilder::start_message_rollback_duration.
readCompacted	✅	✅	ConsumerBuilder::read_compacted.
forceTopicCreation	✅	✅	ConsumerBuilder::force_topic_creation.
Dead-letter policy	✅	✅	ConsumerBuilder::dead_letter_policy + Consumer::drain_dead_letter / republish_dead_letters. Each MultiTopicsConsumer::republish_dead_letters call and its PartitionedConsumer alias independently snapshot their children, then aggregate them sequentially into one shared producer destination, stopping on the first child error without rolling back completed children. Concurrent calls are not serialized; per-child results follow the runtime's destructive drain, and the aggregate adds no cross-call deduplication guarantee. PIP-22/58/124/409.
cryptoFailureAction (PIP-4)	✅	✅	Fail / Discard / Consume all wired end-to-end in magnetar-runtime-tokio::consumer::deliver_post_process.
Encryption (PIP-4)	✅	✅	ConsumerBuilder::encryption accepts a MessageDecryptor.
ConsumerInterceptor SPI	✅	✅	magnetar::ConsumerInterceptor + receive_with_interceptors.
close()	✅	✅	Consumer::close().await is the reliable path: it waits for the broker acknowledgement and reports errors. Dropping the final clone additionally stages a best-effort close and wakes the existing driver (ADR-0077); intermediate clone drops do nothing. This is Rust RAII and pulsar-rs migration parity, beyond Java abandonment semantics — it does not imply that Java garbage collection closes consumers.
unsubscribe()	✅	✅	Consumer / multi-topics expose unsubscribe.
forceUnsubscribe (PIP-313)	✅	✅	Wired through CommandUnsubscribe.force.
availablePermits getter	✅	✅	Consumer::available_permits — the REAL decrementing balance since ADR-0101 (issue #414), matching ConsumerBase#getAvailablePermits; a value pinned at the receiver-queue size while nothing arrives is the client-side signature of a wedged broker dispatcher (see docs/consumer-stall-recovery.md).
availableInQueue getter	✅	✅	Consumer::available_in_queue.
hasReceivedAnyMessage getter	✅	✅	Consumer::has_received_any_message.

Partitioned producer

Feature	Java	Magnetar	Notes
Auto partition discovery	✅	✅	PulsarClient::partitions_for_topic + builder.
MessageRoutingMode (RoundRobin / SinglePartition / Custom)	✅	✅	MessageRoutingMode.
Custom MessageRouter trait	✅	✅	MessageRouter trait + message_router(...).
Murmur3 + JavaStringHash hashers	✅	✅	Murmur3HashHasher / JavaStringHashHasher.
TypedMessageBuilder-equivalent on partitioned producer	✅	✅	PartitionedMessageBuilder.
Per-partition stats / lastSequenceId	✅	✅	Aggregated across child producers.
Auto-update partition count (background ticker)	✅	✅	PartitionedProducerBuilder::auto_update_partitions_interval spawns a tokio::time::interval that signals partitions_changed_notify; user drives refresh_partitions(&client) from the signal.

Partitioned consumer

Feature	Java	Magnetar	Notes
Auto partition discovery + one consumer per partition	✅	✅	PulsarClient::partitioned_consumer.
Full ConsumerBuilder knob forwarding	✅	✅	12 knobs forwarded from builder.
Receive / ack / nack / seek / unsubscribe across partitions	✅	✅	All forwarded.
Aggregate DLQ republish across partitions	❌	✅	PartitionedConsumer::republish_dead_letters independently snapshots the partition children, uses one producer destination, and returns their saturating count. Concurrent calls are not serialized and add no aggregate deduplication guarantee.
Auto-update partition count	✅	✅	PartitionedConsumerBuilder::auto_update_partitions_interval mirrors the producer pattern; signal drives refresh_partitions(&client).

Multi-topics consumer

Feature	Java	Magnetar	Notes
Subscribe to N explicit topics under one subscription	✅	✅	MultiTopicsConsumerBuilder::topics.
Receive / ack / nack / seek across all topics	✅	✅	Per-topic forwarding.
negativeAckWithDelay / ackCumulative	✅	✅	Forwarded.
Aggregate DLQ republish	❌	✅	Each MultiTopicsConsumer::republish_dead_letters call independently snapshots the children, traverses them sequentially in topic order, shares one producer destination, and stops at the first error without rollback. Additions do not enter an existing snapshot, but remove_topic may close a snapshotted shared child handle. Concurrent calls are not serialized and add no aggregate deduplication guarantee.
Dynamic add_topic / remove_topic	✅	✅	MultiTopicsConsumer::add_topic / remove_topic — subscribe / unsubscribe at runtime.
Auto-update partition count (background ticker)	✅	✅	MultiTopicsConsumerBuilder::auto_update_partitions_interval spawns a tokio::time::interval that signals partitions_changed_notify; user drives refresh_partitions(&client) + add_topic(...) from the signal.

Pattern consumer (PIP-145)

Feature	Java	Magnetar	Notes
Regex topic subscription	✅	✅	PatternConsumerBuilder::pattern.
TopicListChanged delta stream	✅	✅	Client::next_topic_list_change.
Manual update() reconcile	✅	✅	PatternConsumer::update(&client) returns a ReconcileReport.
Auto-update background ticker	✅	✅	PatternConsumer::start_auto_reconcile(client, interval) spawns a tokio::time::interval loop that calls update(&client) on every tick; returns a JoinHandle for clean shutdown.

Reader

Feature	Java	Magnetar	Notes
Non-durable exclusive subscription	✅	✅	ReaderBuilder builds on ConsumerBuilder.
startMessageId (Earliest / Latest / explicit)	✅	✅	ReaderBuilder::start_message_id.
startMessageIdInclusive rollback duration	✅	✅	ReaderBuilder::start_message_rollback_duration.
readCompacted	✅	✅	ReaderBuilder::read_compacted.
cryptoKeyReader (PIP-4 decryptor)	✅	✅	ReaderBuilder::encryption.
hasMessageAvailable / seek	✅	✅	Via the underlying consumer surface.
Stats / closure getters (isClosed, etc.)	✅	✅	Reader::is_closed, available_in_queue, available_permits.

Table view

Feature	Java	Magnetar	Notes
Compacted-topic snapshot keyed by message key	✅	✅	TableView::get / for_each / snapshot / keys / values.
Listener registration	✅	✅	TableView::listen (TableViewListener).
Schema-aware TypedTableView	✅	✅	TypedTableView<S> decodes per-read.
startMessageId / subscriptionProperty / property knobs	✅	✅	TableViewBuilder knob set.
Auto-update-partitions ticker	✅	✅	TableViewBuilder::auto_update_partitions_interval(Duration) spawns a background timer that signals TableView::partitions_changed_notify; callers drive refresh_partitions(&client) from the signal.
cryptoKeyReader wired through	✅	✅	TableViewBuilder::encryption + TypedTableViewBuilder::encryption stamp the decryptor onto the underlying ConsumerBuilder.

Transactions (PIP-31)

Feature	Java	Magnetar	Notes
Transaction coordinator client	✅	✅	magnetar-proto::txn::TxnClient.
Begin / commit / abort	✅	✅	Client::new_txn + Transaction::commit / abort.
ADD_PARTITION_TO_TXN / ADD_SUBSCRIPTION_TO_TXN	✅	✅	Client::add_partition_to_txn / add_subscription_to_txn.
Producer publish under txn	✅	✅	OutgoingMessage::txn.
Consumer ack under txn (individual + cumulative + batch)	✅	✅	Consumer::ack_with_txn and friends.
END_TXN_ON_PARTITION / _ON_SUBSCRIPTION cleanup	✅	✅	Driven by end_txn.

Auth + TLS

Feature	Java	Magnetar	Notes
Token auth	✅	✅	magnetar_proto::auth::TokenAuth.
mTLS	✅	✅	magnetar_proto::auth::TlsAuth + tls_trust_certs_pem / tls_trust_certs_file_path.
OAuth2 ClientCredentialsFlow	✅	✅	magnetar_auth_oauth2::ClientCredentialsFlow — POSTs grant_type=client_credentials to the IDP, caches the JWT, refreshes within 30 s of expiry. Reports auth_method_name = "token".
SASL PLAIN (RFC 4616)	✅	✅	magnetar_auth_sasl::SaslPlain — \0<username>\0<password> payload.
SASL Kerberos / GSSAPI	✅	✅	magnetar_auth_sasl::SaslKerberos runs the GSSAPI initiate loop via libgssapi (façade feature auth-sasl-kerberos). The multi-round AUTH_CHALLENGE / AUTH_RESPONSE exchange threads through AuthProvider::respond_to_challenge; the four sans-io test layers per ADR-0024 drive a magnetar_auth_sasl::ScriptedGssapiClient so they stay free of a libkrb5 build dep. End-to-end coverage uses a Dockerised KDC fixture. See ADR-0029.
Athenz (pre-fetched role token)	✅	✅	AthenzProvider::with_role_token — bypass the ZTS round-trip when the caller already holds a valid role token.
Athenz (ZTS round-trip)	✅	✅	feature = "auth-athenz-zts" (default off). The pluggable zts::ZtsClient trait (zts::HttpZtsClient does the reqwest-backed POST) exchanges a signed JWT for a role token; AthenzProvider owns the expiry-aware cache and ensure_role_token(now) / needs_refresh(now) (sans-io clock injection). Build via AthenzProvider::with_default_signer(config) (cfg-active in-tree signer) or AthenzProvider::builder() (custom signer / client / wall_clock). The concrete zts::JwtSigner ships in two flavours — jwt_signer::AwsLcRsSigner and jwt_signer::RingSigner — gated on the crypto-provider matrix per ADR-0035; parsed PKCS#8 DER wrapped in zeroize::Zeroizing<…>, byte-identical deterministic RS256 (RFC 8017 §8.2). Full four-layer cross-runtime coverage (tokio/moonpool/differential + e2e) per ADR-0024. See ADR-0041.
In-band AUTH_CHALLENGE refresh (PIP-30 / PIP-292)	✅	✅	Driver consults the configured AuthProvider and submits CommandAuthResponse.
pulsar+ssl:// URLs	✅	✅	Built-in.
Binary proxy (proxy_to_broker_url)	✅	✅	ClientBuilder::proxy_to_broker_url. Both engines ship the full ProxyConnectionPool — tokio in magnetar-runtime-tokio::pool, moonpool in magnetar-runtime-moonpool::pool (see the ADR-0039 2026-06-01 amendment).

Encryption (PIP-4)

Feature	Java	Magnetar	Notes
MessageEncryptor trait on producer	✅	✅	ProducerBuilder::encryption.
MessageDecryptor trait on consumer	✅	✅	ConsumerBuilder::encryption.
AES-GCM via aws-lc-rs	n/a (Java uses BouncyCastle)	✅	magnetar-messagecrypto::MessageCrypto.
cryptoFailureAction	✅	✅	Fail / Discard / Consume all wired end-to-end in magnetar-runtime-tokio::consumer::deliver_post_process.

Schemas

Feature	Java	Magnetar	Notes
BytesSchema	✅	✅
StringSchema	✅	✅
JsonSchema	✅	✅	Canonicalised via the Avro parser per Codex Q4.
AvroSchema	✅	✅	apache-avro 0.21 — canonical-parsing form.
ProtobufSchema (descriptor)	✅	✅
ProtobufNativeSchema	✅	✅	Byte-identical Java FileDescriptorSet output.
KeyValueSchema	✅	✅	Byte-identical canonical JSON wrapper.
AutoConsumeSchema (broker lookup)	✅	✅	TypedConsumer::receive auto-fetches the broker schema on first call via Connection::get_schema; the result is cached on the schema's Arc<Mutex<Option<pb::Schema>>>.
AutoProduceBytesSchema	✅	✅	TypedProducer::send warms the broker schema on first send via Producer::get_schema; encode() stays pass-through per Java parity.
Int8 / Int16 / Int32 / Int64 / Float / Double / Bool	✅	✅
Date / Time / Timestamp / LocalDate / LocalTime / Instant / LocalDateTime	✅	✅
Schema-version negotiation	✅	✅	Sent on CommandProducer / CommandSubscribe.

Client builder

Feature	Java	Magnetar	Notes
serviceUrl	✅	✅	ClientBuilder::service_url. Broker URLs returned by DIRECT lookups use one shared authority normalizer: Pulsar schemes are ASCII-case-insensitive, explicit ports win, recognized schemes supply 6650/6651, and a scheme-less host inherits the bootstrap scheme's default before resolver dispatch (ADR-0091).
clientVersion	✅	✅	ClientBuilder::client_version.
keepAliveInterval	✅	✅	ClientBuilder::keepalive.
operationTimeout	✅	✅	ClientBuilder::operation_timeout is one total setup-operation deadline spanning partition metadata, PIP-145 topic-list snapshots, lookup/redirect dialing, retry backoff, producer-open, subscribe acknowledgement, and every child of a composite builder. The same operation context preserves the newest retryable broker error if a later stage reaches the deadline.
Operation retry policy	✅	✅	ClientBuilder::operation_retry(OperationRetryConfig) configures operation-specific transient errors, exponential backoff, and the number of retries after the initial attempt. Producer-open additionally retries both producer-quota variants and ProducerBusy; subscribe additionally retries ConsumerBusy. Provisional attachment retries re-run lookup and routing with a fresh handle, while established reattachment remains driver-owned. The default is 2 s initial / 8 s maximum backoff with eight retries; None removes the count cap but remains bounded by operationTimeout (ADR-0080, issue #343).
maxMessageSize	✅	✅	ClientBuilder::max_message_size.
tlsTrustCertsFilePath	✅	✅	ClientBuilder::tls_trust_certs_file_path.
tlsAllowInsecureConnection	✅	✅	ClientBuilder::tls_allow_insecure_connection(true) — accepts any server cert via a custom rustls verifier. Insecure, do not use in production.
enableTlsHostnameVerification	✅	✅	ClientBuilder::tls_hostname_verification_enable(bool) — true uses the standard WebPKI verifier; false paired with tls_trust_certs_pem routes through magnetar_runtime_tokio::tls_config_no_hostname which delegates chain check to WebPKI and intercepts only NotValidForName.
serviceUrlProvider (URL rotation)	✅	✅	ClientBuilder::service_url_provider(Arc<dyn ServiceUrlProvider>) — the supervised reconnect path calls provider.get_service_url() on every reconnect attempt so cluster-failover policies can swap URLs between attempts.
proxyServiceUrl (binary proxy)	✅	✅	ClientBuilder::proxy_to_broker_url. Both engines route proxy lookups through a per-broker ProxyConnectionPool (ADR-0039, moonpool amendment 2026-06-01).
Authentication plugin	✅	✅	ClientBuilder::auth(Arc<dyn AuthProvider>).
memoryLimit	✅	✅	ClientBuilder::memory_limit(bytes, MemoryLimitPolicy). Both FailImmediately (atomic CAS, ADR-0017) and ProducerBlock (Waker slab, ADR-0020) ship.
dnsResolver customisation	✅	✅	ClientBuilder::dns_resolver(Arc<dyn DnsResolver>) — Transport::connect_with_resolver resolves via the provider on every (re)connect; TokioDnsResolver is the default.
connectionsPerBroker	✅	✅	ClientBuilder::connections_per_broker(n) (default 1; 0/1 ≡ Java's one-connection floor) — opens up to n connections per broker and round-robins producers AND consumers across them, so a logical producer fleet spreads its publish load over several connections instead of contending on one (ADR-0073, issue #314). Deterministic round-robin (not Java's random key) keeps both engines bit-reproducible.
isClosed / shutdown / getLastDisconnectedTimestamp	✅	✅	All exposed on PulsarClient.
Cluster failover (PIP-121)	✅	✅	ServiceUrlProvider + StaticServiceUrlProvider + ControlledClusterFailover (proto) + AutoClusterFailover (runtime, with user-supplied HealthProbe callback + background tokio task). All three plug into ClientBuilder::service_url_provider.

Open structural gaps

• Moonpool compression parity. Java parity is fully satisfied by the tokio engine (ADR-0019). The façade's producer, consumer, reader, partitioned, multi-topics, pattern, TableView, transaction, and typed-schema surfaces are engine-generic, but the Moonpool producer still rejects non-None compression. Per-feature, per-engine status lives in the parity matrix above, with runtime details in docs/moonpool-engine.md.
• PIP-460 scalable topics ship as an experimental surface behind feature = "scalable-topics" (default off), speaking the wire protocol vendored from Apache Pulsar 5.0.0-M1. M1 is a milestone rather than a GA release, so the surface may still move before Pulsar 5.0 final; against a Pulsar 4.x broker the capability is negotiated away and no scalable-topic command is emitted.
• PIP-466 V5 surface ships as an experimental, engine-generic wrapper behind feature = "experimental-v5-client" (default off). No wire change — it wraps the v4 surface.
• SASL ships both mechanisms end-to-end: PLAIN (RFC 4616) under the default auth-sasl feature, and Kerberos/GSSAPI via libgssapi under the auth-sasl-kerberos feature. The multi-round AUTH_CHALLENGE exchange threads through AuthProvider::respond_to_challenge. The four sans-io test layers drive a deterministic ScriptedGssapiClient; the e2e layer runs against a Dockerised KDC. See ADR-0029.
• Athenz ships both the pre-fetched role-token path and the opt-in ZTS round-trip (auth-athenz-zts). Production-style ZMS+ZTS+certificate bootstrap remains out of scope for the local fixture.

---

Supported PIPs

PIP	Title	Status	Lives in
PIP-4	End-to-end encryption (AES-GCM)	✅	magnetar-messagecrypto, crypto_bridge in magnetar
PIP-22	DLQ topic	✅	ConsumerBuilder::dead_letter_policy
PIP-30	In-band AUTH_CHALLENGE refresh	✅	magnetar-proto::auth, driver
PIP-31	Transactions	✅	magnetar-proto::txn, Client::new_txn
PIP-37	Chunking + ack-timeout redelivery backoff	✅	magnetar-proto::producer, trackers::nack; consumer-side reassembly bounded in magnetar-proto::consumer (max_pending_chunked_message cap 10, expire_time_of_incomplete_chunked_message 60s, auto_ack_oldest_chunked_message_on_queue_full false — Java-matching), with accepted incomplete chunks repaying their broker permits before logical-message reassembly completes (ADR-0076)
PIP-54	Partial-batch ACK (ack_set bitset)	✅	magnetar-proto::consumer
PIP-58	Retry-letter topic	✅	Consumer::reconsume_later
PIP-68	Exclusive producer access mode	✅	ProducerBuilder::access_mode
PIP-90	Broker-entry metadata envelope	✅	magnetar-proto::frame (magic 0x0e02), IncomingMessage::broker_*
PIP-124	Multi-DLQ topics for KeyShared	✅	DLQ policy infra
PIP-145	Topic list watcher (regex pattern)	✅	magnetar-proto::topic_watcher, PatternConsumer
PIP-292	Better in-band auth refresh ergonomics	✅	Driver event handler
PIP-313	Force unsubscribe	✅	CommandUnsubscribe.force plumbed
PIP-34 / 119 / 282 / 379	Key_Shared family	✅	KeySharedConfig + builder
PIP-409	DLQ + retry-letter polish	✅	DLQ + reconsume_later wiring
PIP-391	Batch-index ACK polish	✅	Pairs with PIP-54
PIP-188	TOPIC_MIGRATED	✅	Wire opcode decoded; tokio driver's event loop catches ConnectionEvent::TopicMigrated, logs the new-broker hint, and returns an error from driver_loop_inner so the supervisor triggers Connection::reset + reconnect. rebuild_producers / rebuild_consumers re-attach every still-open handle on the new socket.
local	Anti-thrash policy (ADR-0028)	✅ (opt-in)	Per-handle ack-then-drop detector + connection-level cooldown. Mitigates broker-driven post-restart cascades (Pulsar PR #14467 / #13428 / #12846 — ServerCnx#handleProducer ↔ AbstractTopic#addProducer race). SupervisorConfig::anti_thrash_threshold default None.
PIP-460	Scalable topics	🟡	Experimental, behind feature = "scalable-topics" (default off), ADR-0093 (supersedes ADR-0031). Speaks the surface vendored from Apache Pulsar 5.0.0-M1 — a milestone, not GA, so it may still move before 5.0 final. Ships the topic:// URL scheme, the upstream lookup / update / close commands as ordinary BaseCommand frames (the lookup doubles as the layout subscribe, keyed by a client-allocated session_id), the DagWatchSession sans-io state machine (whole-layout snapshots ordered by a monotonic epoch, split/merge derived from parent_ids/child_ids edges), both-engine ScalableTopicsApi impls, magnetar::scalable::StreamConsumer (StreamConsumer-only, drops on DAG change), and the magnetarctl topic-info CLI. Pulsar 4.x compatible: the capability is negotiated via FeatureFlags.supports_scalable_topics on the handshake and no scalable command is written to a peer that did not advertise it. MessageId is not extended — M1 carries no segment field on MessageIdData. A layout whose epoch does not advance is ignored rather than fatal — the broker re-sends its snapshot on the watch the lookup opens, and treating that as a violation ended the watch (ADR-0095). Consumer registration (CommandScalableTopicSubscribe → ConsumerAssignment, rebalances surfaced as AssignmentDelta), the namespace-level topic watch and PIP-473 transaction-coordinator discovery all ship, each negotiated on its own feature flag. QueueConsumer / CheckpointConsumer, per-segment message fan-out, controller-election awareness and repartition remain out of scope.
PIP-466	V5 client API surface	✅	Behind feature = "experimental-v5-client" (default off). Engine-generic per ADR-0032. magnetar::v5 exposes PulsarClientV5<E: Engine = TokioEngine> (with v4() escape hatch), v5::Producer<E>, v5::StreamConsumer<E> (Exclusive / Failover), v5::QueueConsumer<E> (Shared / KeyShared), and the v5::mapping field-translation table. Moonpool callers name PulsarClientV5<MoonpoolEngine<P>> directly. Wraps the v4 surface — no wire change. See ADR-0032.
PIP-180	Shadow topic	✅	Admin REST (create_shadow_topic / delete_shadow_topic / get_shadow_topics / get_shadow_source), producer-side send_with_source_message_id propagating CommandSend.message_id, consumer-side MessageReceivedFromShadow event, structural MessageId equality across source ⇄ shadow. See ADR-0033.
PIP-415	getMessageIdByIndex	✅	magnetar-admin::AdminClient::topic_get_message_id_by_index — REST-only per PIP-415 (binary-protocol section intentionally empty; canonical implementation apache/pulsar#24222 is admin / broker / CLI only)
PIP-33	Replicated subscriptions	✅	ConsumerBuilder::replicate_subscription_state(bool) flips CommandSubscribe field 14; receive-path filter in magnetar-proto::conn drops REPLICATED_SUBSCRIPTION_* markers and surfaces them via PulsarClient::next_replicated_subscription_marker / poll_replicated_subscription_marker. Client never originates markers — broker-side machinery only. See ADR-0034.
PIP-121	Cluster failover (Auto + Controlled)	✅	ServiceUrlProvider + StaticServiceUrlProvider + ControlledClusterFailover (proto) + AutoClusterFailover (runtime with HealthProbe). Active URL re-resolved on every supervised-reconnect attempt.

---

Engine-by-engine surface coverage

Magnetar publishes two engines that drive the same sans-io state machine. Pick at compile time via feature flags. The parity matrix above is the per-feature, per-engine status snapshot; this section describes each engine's surface.

magnetar-runtime-tokio — production (default)

• TLS via tokio-rustls (ring backend); no native-tls, no openssl.
• One driver task per connection — see ARCHITECTURE.md §"The driver loop".
• The user-facing futures (Consumer::receive, Producer::send, …) lock the shared state machine, register their Waker in a slab, and wait. The driver picks them up as the matching OpOutcome lands.
• This is what magnetar::PulsarClient wires by default (PulsarClient<TokioEngine>).

magnetar-runtime-moonpool — deterministic simulation

• Drives the same sans-io state machine as the tokio engine over moonpool_core::Providers (a bundle of NetworkProvider, TimeProvider, TaskProvider, RandomProvider, StorageProvider). Plug TokioProviders for production-style runs against a real broker, or moonpool_sim::SimProviders for Moonpool 0.8's native seeded executor, virtual clock, simulated network/storage, and reproducible chaos without an ambient Tokio runtime.
• Provider-generic tasks, timers, and concurrent waits use TaskProvider, TimeProvider, and moonpool_core::select!; simulation observability uses named flat tracing events queried through TraceQuery (ADR-0078).
• TLS uses a local rustls::ClientConnection adapter (tls.rs) that drives read_tls / process_new_packets / write_tls over the moonpool byte pipe — the handshake stays deterministic under chaos.
• See docs/moonpool-engine.md for the engine's surface, supervised reconnect, chaos test pack, and the tokio ↔ moonpool differential equivalence harness.

---

Supported broker versions

• Pulsar 4.0+ (LTS). The CONNECT frame advertises ProtocolVersion::V21 and the connection falls back to whichever lower version the broker reports on CONNECTED.
• The end-to-end suite runs against apachepulsar/pulsar:4.0.4.

---

Status

Magnetar targets a feature-complete Apache Pulsar driver with full Java-client parity over a shared sans-io protocol state machine (ADR-0010, ADR-0019). Java parity is satisfied by the production tokio engine. The deterministic-simulation Moonpool engine shares the generic façade and differential EventStream contract, with the compression limitation documented above.

The bulk of the parity matrix above ships on main, including:

• PIP-180 shadow topic (ADR-0033).
• PIP-33 replicated subscriptions (ADR-0034).
• SASL Kerberos / GSSAPI (ADR-0029).
• Pluggable rustls crypto provider (aws-lc-rs / ring / openssl / fips — ADR-0035).
• Daily 128-random-seed moonpool sweep (ADR-0036).
• Swarm-tested simulation configurations — each seed runs a derived subset of buggify labels + workload operations, with a reserved inclusive slice (ADR-0097).
• Anti-thrash supervised reconnect policy (opt-in, ADR-0028).

Known open work is narrow and tracked in docs/follow-ups.md.

The public API is stable as of 1.0.0 and follows Semantic Versioning; the experimental surfaces noted above (PIP-460 scalable topics, the CLI produce/consume subcommands) are excluded from that guarantee.

---

Validation

The whole workspace builds against stable Rust 1.91 (ADR-0079).

# Build / lint / format
cargo build --workspace --all-features
cargo clippy --workspace --all-features -- -D warnings
cargo +nightly fmt --check

# Unit + integration tests (no broker needed)
cargo test --workspace

# Dependency audits
cargo deny check

# Docs
RUSTDOCFLAGS="-D warnings" \
  cargo doc --workspace --all-features --no-deps

End-to-end tests against a real broker run as part of cargo test --workspace --all-features (ADR-0046 — no --features e2e, no #[ignore]); per-PR CI parallelizes that surface into one non-e2e cell and four e2e cells (ADR-0098). Docker is the only prerequisite; the suite spins pulsar:4.0.4 via testcontainers-rs.

Additional xtask checks specific to the sans-io invariants:

cargo run -p xtask -- check-no-channels   # greps src/** for banned channel crates
cargo run -p xtask -- check-no-io-deps    # magnetar-proto must not depend on any I/O crate
cargo run -p xtask -- codegen --check     # asserts proto codegen has no drift

---

License

Apache-2.0 — see LICENSE and NOTICE. The project vendors a verbatim copy of the Apache Pulsar wire protocol definition (PulsarApi.proto, PulsarMarkers.proto), released by the Apache Software Foundation under Apache-2.0.

See GUIDELINES.md and CONTRIBUTING.md for project conventions before sending a patch.