Struct ClientBuilder 

pub struct ClientBuilder { /* private fields */ }

Builder for PulsarClient.

Implementations

impl ClientBuilder

pub fn service_url(self, url: impl Into<String>) -> Self

Set the Pulsar service URL (pulsar:// or pulsar+ssl://).

pub fn dns_resolver(self, resolver: Arc<dyn DnsResolver>) -> Self

Plug in a custom DNS resolver. Mirrors Java ClientBuilder#dnsResolver. Used on every connection attempt (initial + reconnect) instead of tokio’s default tokio::net::lookup_host. Useful for service-mesh sidecar resolution, IPv4/IPv6 preference, pinning, etc.

Default: tokio’s built-in DNS via magnetar_runtime_tokio::TokioDnsResolver.

pub fn memory_limit(self, bytes: usize, policy: MemoryLimitPolicy) -> Self

Set the global publish memory budget for the client. Mirrors Java ClientBuilder#memoryLimit(long, MemoryLimitPolicy). bytes = 0 disables the limit (matches Java default).

Enforcement: under MemoryLimitPolicy::FailImmediately, every Producer::send reserves the payload bytes against the budget via an AtomicU64 CAS loop on ConnectionShared::memory_used BEFORE the payload reaches the sans-io state machine. Sends that would push past the limit are rejected synchronously with magnetar_runtime_tokio::ClientError::MemoryLimitExceeded. The reservation is released on SendFut completion (success or error) and on cancellation (via Drop).

Under MemoryLimitPolicy::ProducerBlock, the send future parks on a Notify-based wait until the budget frees up — both engines (TokioEngine, MoonpoolEngine<P>) implement this policy; see docs/memory-limit.md.

pub fn connections_per_broker(self, n: usize) -> Self

Set the number of connections the client opens to each broker. Mirrors Java ClientBuilder#connectionsPerBroker(int) (issue #314, ADR-0073).

Default (and 0/1): one connection per broker — every producer and consumer for a given broker shares a single TCP connection, exactly as before. With n > 1, the client opens up to n connections per broker and round-robins producers / consumers across them, so a single logical producer fleet can spread its publish load over several independent connections instead of contending on one (the per-connection driver, its send path, and its receipt-read path are independent per connection). This removes the send-side back-pressure that otherwise forces applications to hand-roll a pool of PulsarClients.

0 is treated as 1 (matching Java, where the floor is one connection).

pub fn service_url_provider(self, provider: Arc<dyn ServiceUrlProvider>) -> Self

Set a pluggable magnetar_proto::ServiceUrlProvider consulted on every (re)connection attempt. Mirrors Java ClientBuilder#serviceUrlProvider(ServiceUrlProvider) — lays the groundwork for PIP-121 cluster failover (AutoClusterFailover / ControlledClusterFailover). When set, the provider’s get_service_url() is used at connect time; the unset form retains the legacy service_url(...) shortcut and is internally wrapped in a magnetar_proto::StaticServiceUrlProvider at build time.

pub fn client_version(self, version: impl Into<String>) -> Self

Override the advertised client version.

pub fn keepalive(self, dur: Duration) -> Self

Set the keep-alive (ping) interval.

pub fn stats_interval(self, dur: Duration) -> Self

Set the cadence at which the client re-samples every producer’s and consumer’s rolling rate window — the sampling that makes magnetar_proto::ProducerStats::msgs_per_sec / bytes_per_sec and their magnetar_proto::ConsumerStats counterparts nonzero. Mirrors Java ClientBuilder#statsInterval(long, TimeUnit).

The tick runs inside the sans-io state machine’s existing poll_timeout / handle_timeout deadline loop (ADR-0089), so it applies to every producer and consumer on the client — including the per-partition and per-topic children behind crate::PartitionedProducer, crate::MultiTopicsConsumer and crate::PatternConsumer, whose aggregate_stats() folds therefore sum real rates rather than zeros. There is deliberately no per-wrapper fan-out method: Java’s wrappers have none either, and one clock ticking every child is what makes the folded sum well-defined.

Duration::ZERO disables the sweep, spelling Java’s statsIntervalSeconds = 0. Leaving the knob unset inherits ConnectionConfig::stats_interval’s default.

A producer or consumer created mid-window has no baseline yet, so its first sweep only seeds one and it reports 0.0 for one further interval. Java behaves identically.

pub fn consumer_stall_timeout(self, dur: Duration) -> Self

Arm the per-consumer stall watchdog (issue #414).

A consumer that holds un-spent broker permits over an empty receive queue, in a dispatch-eligible state, for dur without a single dispatch unit arriving surfaces one warn! and one ConnectionEvent::ConsumerStalled — exactly one per stall episode, re-armed by the next dispatch. That is its only effect unless Self::consumer_stall_auto_recovery is also set; otherwise recovery stays an explicit call to Consumer::resubscribe(), escalating to an operator-side pulsar-admin topics unload for a dispatcher-wide broker fault.

This is the one silence the ADR-0058 connection keepalive cannot see: a broker whose dispatcher has wedged for ONE subscription keeps answering PING with PONG.

Unset by default — the mechanism ships disarmed, since an armed deadline perturbs the moonpool engine’s simulated wake schedule even when it never fires, and Java has no per-consumer dispatch watchdog to inherit a parity value from. Duration::from_secs(30) is the recommended production value: it matches the keepalive and ack-response cadences, and is far longer than any legitimate dispatch gap on a subscription that holds permits over an empty queue.

Duration::ZERO disables it explicitly, mirroring how Self::stats_interval spells its disable.

pub fn consumer_stall_auto_recovery(self, max_attempts: u32) -> Self

Let the stall watchdog recover a wedged consumer by itself, at most max_attempts times per stall streak (issue #414, ADR-0103).

Each attempt is the same in-place re-attach Consumer::resubscribe() performs — zero this client’s permit mirrors, fail the orphaned in-flight acks, re-emit CommandSubscribe for the same consumer id on the live connection, and let the broker’s Success release a fresh initial CommandFlow. No transport reconnect, no other consumer or producer disturbed, and the receiver queue left intact.

The ConsumerStalled event and its warn! are emitted either way, so arming this adds a recovery attempt without ever hiding the diagnosis.

Requires Self::consumer_stall_timeout — with no window there is no stall episode, and this knob is inert. Unset by default, and 0 disables it explicitly, mirroring how Self::consumer_stall_timeout spells its disable.

Choosing the bound

At most one attempt is made per stall episode and an episode closes at most once per consumer_stall_timeout, so max_attempts caps a sequence already limited to one re-subscribe per window: with a 30 s window, 3 spends at most three re-subscribes over ninety seconds before giving up. The counter resets on real progress only — one broker dispatch unit actually arriving — so a consumer that recovers and later wedges again gets its full budget back, while a consumer the broker acks but never dispatches to does not.

Keep it small. An attempt repairs this client’s own slot in the broker’s dispatcher and lifts the subscription’s aggregate permit counter by one receiver-queue window; issue #414’s production failure was dispatcher-WIDE, with that aggregate observed at -177300, which no realistic number of re-subscribes reaches. When the budget is exhausted the client stops and logs the escalation — pulsar-admin topics unload — instead of re-subscribing forever against a fault it cannot repair. See docs/consumer-stall-recovery.md.

pub fn operation_timeout(self, dur: Duration) -> Self

Set the total deadline for one broker-facing setup operation.

The budget includes partition metadata, topic-list snapshots, lookup and redirect routing, retry backoff, producer-open or subscribe attachment, and every child of a composite builder. The operation preserves the newest retryable broker diagnostic so a later deadline returns it instead of a generic timeout.

pub fn operation_retry(self, config: OperationRetryConfig) -> Self

Configure broker-operation retries independently from transport reconnection.

Applies to lookup, partition metadata, producer-open, and subscribe. Producer-open additionally retries both producer-quota variants and ProducerBusy; subscribe additionally retries ConsumerBusy. Before first attachment, producer-open and subscribe retries re-run lookup and routing with a fresh provisional handle. Established reattachment remains driver-owned. max_retries counts re-issues after the initial attempt; None removes the count cap but the enclosing Self::operation_timeout deadline still bounds the operation.

pub fn ack_response_timeout(self, timeout: Duration) -> Self

Bound how long the client waits for a CommandAckResponse after issuing a CommandAck. In-flight acks past enqueued_at + timeout resolve with a synthetic broker error carrying code=-1, message="ack timeout" on the next state-machine tick — mirrors crate::ProducerBuilder::send_timeout’s shape and rationale.

The default is 30 s (mirrors the send_timeout Java-parity default, ADR-0072), so an ack whose response is lost or dropped in flight fails deterministically rather than hanging the caller’s ack().await forever. A same-broker CloseConsumer (bundle reassignment, issue #307) additionally fails every ack pending against the torn-down consumer id immediately, ahead of this deadline — this knob is the generic backstop for every other cause of a dropped response. Call Self::disable_ack_response_timeout for the unbounded (never-times-out) behavior.

pub fn disable_ack_response_timeout(self) -> Self

Disable the ack-response timeout: in-flight acks never resolve with a synthetic timeout error — they wait indefinitely for the broker’s CommandAckResponse (or a session-loss / terminal error, or the same-broker CloseConsumer orphan sweep, which is unaffected by this knob). Overrides the 30 s default.

pub fn max_message_size(self, size: usize) -> Self

Override the default max_message_size used as the chunking threshold when the broker does not advertise one on CommandConnected. The Pulsar default is 5 MiB; match the broker’s configured maxMessageSize to avoid mis-sized chunks. Mirrors Java ClientBuilder#maxMessageSize.

pub fn proxy_to_broker_url(self, url: impl Into<String>) -> Self

Set the proxy-to-broker URL for the binary proxy path. The connection then opens against the proxy with the broker URL stamped on the CommandConnect.proxy_to_broker_url field. Mirrors Java ClientBuilder#proxyServiceUrl(... ProxyProtocol.SNI). Leave unset for direct broker connections.

pub fn enable_reconnect(self, config: SupervisorConfig) -> Self

Enable the auto-reconnect supervisor with the supplied magnetar_proto::SupervisorConfig. When set, runtime engines wrap the driver loop in a magnetar_proto::Backoff-driven reconnect cycle so the connection survives transport failures. Without this knob the driver exits on the first I/O error (matches the pre-supervisor behavior). Mirrors Java’s PulsarClientImpl reconnect loop.

Note: pending in-flight producer/consumer requests issued before the drop surface a “session lost” outcome on the new connection; transparent re-subscription and producer reattachment across reconnects is a future enhancement layered on top of this scaffold.

pub fn auth(self, provider: Arc<dyn AuthProvider>) -> Self

Use the supplied auth provider to populate the initial CONNECT auth data, and keep the provider for in-band CommandAuthChallenge refresh (PIP-30 / PIP-292).

BREAKING CHANGE: the provider’s magnetar_proto::AuthProvider::initial is now invoked inside Self::build and any error it returns surfaces through PulsarError::Config — the previous behaviour silently dropped the error via .ok(), which would have let an uncached OAuth2 flow / a missing token file / an expired credential open an anonymous connection (CWE-287). Callers using a provider whose initial() returns Err(AuthError::Invalid) until an out-of-band warm-up runs (e.g. OAuth2Provider::ensure_fresh) MUST warm the provider before calling Self::build.

pub fn tls_trust_certs_pem(self, pem: impl Into<Vec<u8>>) -> Self

Mirrors Java ClientBuilder#tlsTrustCertsFilePath (PEM-supplied equivalent — magnetar keeps the façade I/O-free, callers read the file themselves via std::fs::read(path)? and pass the bytes). Supplies a PEM-encoded chain (typically a self-signed CA used by the broker). When set, the connection’s TLS handshake validates the broker against this chain INSTEAD OF the system trust store. Only honoured for pulsar+ssl:// URLs.

pub fn tls_allow_insecure_connection(self, on: bool) -> Self

Mirror of Java ClientBuilder#tlsAllowInsecureConnection. When true, the TLS handshake accepts any server certificate without verifying its trust chain — useful for local development against a self-signed broker or for CI / e2e against an ephemeral container. Insecure for production: the client cannot tell a real broker from a MITM.

Default: false. Only honoured for pulsar+ssl:// URLs. Overrides any tls_trust_certs_pem chain when set.

pub fn tls_hostname_verification_enable(self, on: bool) -> Self

Mirror of Java ClientBuilder#enableTlsHostnameVerification. When true (the default), the handshake additionally checks the server certificate’s CN / SAN matches the broker hostname from the URL. When false, the chain is still verified but the hostname mismatch is tolerated.

Default: true (matches Java’s secure default). When Self::tls_allow_insecure_connection is true this flag is moot — the verifier already accepts everything.

Note: today only the “off + insecure both true” combination is runtime-enforced via magnetar_runtime_tokio::insecure_tls_config. A hostname-only-skip verifier (chain on, hostname off) is a planned follow-up; passing false without also enabling tls_allow_insecure_connection is currently treated as the default (hostname verification stays on).

pub async fn build(self) -> Result<PulsarClient, PulsarError>

Build and connect the client.

Errors

Returns PulsarError::Config if the service URL is missing, or PulsarError::Client if the underlying tokio engine fails to connect.

Trait Implementations

impl Clone for ClientBuilder

fn clone(&self) -> ClientBuilder

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ClientBuilder

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for ClientBuilder

fn default() -> Self

Returns the “default value” for a type.