Struct ConnectionConfig 

pub struct ConnectionConfig {

    pub client_version: String,
    pub auth_method_name: String,
    pub auth_data: Option<Bytes>,
    pub protocol_version: i32,
    pub feature_flags: FeatureFlags,
    pub keepalive_interval: Duration,
    pub operation_timeout: Duration,
    pub connect_timeout: Duration,
    pub connect_max_retries: u32,
    pub default_compression: CompressionKind,
    pub default_max_message_size: usize,
    pub proxy_to_broker_url: Option<String>,
    pub supervisor: Option<SupervisorConfig>,
    pub memory_limit_bytes: u64,
    pub memory_limit_policy: MemoryLimitPolicy,
    pub max_pending_lookups: usize,
    pub redirect_url_allow_list: Option<RedirectUrlAllowList>,
    pub ack_response_timeout: Option<Duration>,
    pub stats_interval: Option<Duration>,
    pub consumer_stall_timeout: Option<Duration>,
    pub consumer_stall_auto_recovery: Option<u32>,
    pub buggify: Buggify,
}

Connection configuration.

Fields

client_version: String

Client version string advertised in CommandConnect.

auth_method_name: String

Authentication method name (e.g. "none", "token").

auth_data: Option<Bytes>

Initial auth data (when an auth provider already has a token).

protocol_version: i32

Protocol version to advertise; 21 covers Pulsar 4.x.

feature_flags: FeatureFlags

Capabilities to advertise on connect.

keepalive_interval: Duration

Keepalive (ping) interval. Default 30 s.

operation_timeout: Duration

Total timeout for one connect or broker-facing setup operation. Default 30 s.

For producer/consumer setup, one provider-backed timer spans partition metadata, lookup and redirect dialing, operation-retry backoff, and producer-open or subscribe acknowledgement (ADR-0080).

For connect, it also bounds the post-dial CONNECT → CONNECTED handshake: once the dial succeeds, the engines arm a single operation_timeout deadline over the handshake read loop so a broker that accepts the TCP SYN but never replies to CommandConnect surfaces a bounded Io(TimedOut) instead of parking forever. ADR-0052’s dual cap scopes to the dial; this extends the same total budget to the handshake (Java operationTimeoutMs parity). (ADR-0052, ADR-0080)

connect_timeout: Duration

Per-attempt timeout for the initial TCP/TLS dial. A dial that does not complete within this budget is abandoned and retried (see Self::connect_max_retries). Mirrors Java’s connectionTimeoutMs. Default 10 s (Java connectionTimeoutMs). (ADR-0052)

connect_max_retries: u32

Bounded retries for the initial dial when it times out or fails with a transient I/O error. 0 means a single attempt (no retry). Each retry re-dials after an exponential backoff. The Pulsar handshake that follows a successful dial is NOT retried here — surviving mid-stream transport drops is the supervisor’s job (Self::supervisor). Default 8.

default_compression: CompressionKind

Default compression for producers (overridable per producer).

default_max_message_size: usize

Default max-message-size if the broker omits it. Pulsar default = 5 MiB.

proxy_to_broker_url: Option<String>

Optional proxy-to-broker URL for the binary proxy path.

supervisor: Option<SupervisorConfig>

Optional auto-reconnect supervisor. When Some, runtime engines wrap the driver loop in a backoff-driven reconnect cycle that survives transport failures. None (the default) keeps the pre-supervisor behavior — driver exits on the first I/O error. Mirrors Java’s PulsarClientImpl reconnect loop.

memory_limit_bytes: u64

Global publish memory budget in bytes. 0 (the default) disables the limit. Runtime engines that honour this enforce a CAS-reserve on every Producer::send before queueing into the sans-io state machine; sends that would push the in-flight bytes past the limit are gated by memory_limit_policy. Mirrors Java ClientBuilder#memoryLimit.

memory_limit_policy: MemoryLimitPolicy

Policy applied when the global publish memory budget is exhausted. Defaults to MemoryLimitPolicy::FailImmediately to match the Java client default. MemoryLimitPolicy::ProducerBlock makes the runtime park the offending send future on a waker slab until enough budget frees up. Ignored when memory_limit_bytes is 0.

max_pending_lookups: usize

Cap on the total number of in-flight broker LOOKUP + partitioned-topic-metadata requests on this connection. 0 (the default) preserves the historical unbounded behaviour and matches Java.

Set to a small positive value (e.g. 1024) to harden the client against pending-lookup memory amplification by a misbehaving or hostile broker — closes the “pending-lookup memory amplification” finding from the lookup multi-agent review. Requests that would exceed the cap surface synchronously as LookupOutcome::Failed { code: 0, message: "lookup rejected: max pending" } without ever touching the wire.

redirect_url_allow_list: Option<RedirectUrlAllowList>

Allow-list for broker-advertised redirect URLs (PIP-188 TopicMigrated, CommandLookupTopicResponse proxy URLs, CommandCloseProducer / CommandCloseConsumer reassignment URLs).

None (the default) is permissive: the runtime trusts every URL the broker advertises and honours it on the next handshake under the same AuthProvider. This preserves pre-allow-list behaviour.

Some(allow_list) is defence-in-depth: the runtime validates every broker-advertised URL through RedirectUrlAllowList::is_allowed before re-dialling. A rejected URL surfaces a ConnectionEvent::RedirectUrlRejected and short-circuits the re-dial; the runtime keeps using the original URL (no credentials are sent to the rejected host).

See RedirectUrlAllowList for the threat-model rationale.

ack_response_timeout: Option<Duration>

Backstop deadline for an in-flight ack (Connection::ack and its grouped/chunk-auto-ack internal callers) that never gets a CommandAckResponse. handle_timeout reaps any ack whose enqueued_at + ack_response_timeout has elapsed with a synthetic code=-1, message="ack timeout" error, mirroring the send_timeout shape (ADR-0072). Default Some(30 s) — Java-parity canonical default (mirrors CreateProducerRequest::send_timeout’s 30 s default, #304). None disables the backstop entirely: no deadline is ever computed by poll_timeout and no spurious wakeups are scheduled (load-bearing for moonpool determinism — an armed deadline that never fires would still perturb the simulated clock’s wake schedule).

This is independent of the issue #346 same-broker CloseConsumer sweep, which fails orphaned acks immediately rather than waiting out this deadline; the backstop only matters when the broker goes silent without ever tearing the consumer down (e.g. a dropped CommandAckResponse on an otherwise healthy connection).

stats_interval: Option<Duration>

Cadence at which handle_timeout re-samples every producer’s and consumer’s rolling rate window (ProducerState::record_rate_window / ConsumerState::record_rate_window), which is what makes ProducerStats::msgs_per_sec / bytes_per_sec and their ConsumerStats counterparts nonzero. Mirrors Java ClientConfigurationData.statsIntervalSeconds, whose recorders self-tick on the client-wide HashedWheelTimer; magnetar expresses the same obligation as a deadline on the existing poll_timeout / handle_timeout loop instead of a task (ADR-0089).

Default Some(60 s) — the Java-parity canonical value. None disables the sweep entirely: no deadline is ever computed by poll_timeout and no spurious wakeups are scheduled (load-bearing for moonpool determinism — an armed deadline that never fires would still perturb the simulated clock’s wake schedule, exactly as documented for Self::ack_response_timeout above). Java spells the same disable as statsIntervalSeconds = 0, which ClientBuilder::stats_interval(Duration::ZERO) maps onto.

The per-slot baseline is each slot’s existing last_rate_snapshot timestamp, so a slot created mid-window has none yet: its first sweep only seeds a baseline and it reports 0.0 for one further interval. Java’s recorders have the identical property (a recorder constructed mid-window publishes its first rate one statsIntervalSeconds later), so this is parity-correct rather than a rounding artefact.

consumer_stall_timeout: Option<Duration>

Per-consumer stall watchdog window (issue #414). handle_timeout surfaces one ConsumerStalled event for a consumer that has held un-spent broker permits over an empty receive queue, in a dispatch-eligible state, for this long without a single dispatch unit arriving — and exactly one per stall episode, re-armed by the next dispatch. See ConsumerState::poll_stall.

The connection keepalive of ADR-0058 cannot cover this: a broker whose Shared dispatcher has wedged for ONE subscription keeps answering PING with PONG, so last_activity never ages and no connection-level deadline ever fires. Issue #414 is exactly that shape — survivors of a consumer-churn window receive ~20 messages then nothing, the broker’s own availablePermits for the subscription goes hugely negative, acks_failed stays 0, and the client reports no error at all.

Default None — the mechanism ships disarmed. Two reasons, both precedented here: an armed deadline that never fires still perturbs the moonpool engine’s simulated wake schedule (the rationale Self::ack_response_timeout and Self::stats_interval both carry), and ADR-0089 landed its sweep off for exactly this reason so the flip could be its own bisectable commit with its own seed sweep. There is no Java counterpart forcing a parity value either — the Java client has no per-consumer dispatch watchdog.

Some(Duration::from_secs(30)) is the recommended production value: it matches Self::keepalive_interval and Self::ack_response_timeout, so a stall is reported on roughly the cadence at which every other silence on the connection is already judged, and it is far longer than any legitimate broker dispatch gap on a subscription that holds permits over an empty queue. Shorter windows start reporting ordinary idle backlogs; much longer ones defeat the point of a watchdog.

Emitting the event is the only effect unless Self::consumer_stall_auto_recovery is also set; otherwise recovery stays an explicit Connection::resubscribe_consumer_in_place.

consumer_stall_auto_recovery: Option<u32>

Bounded automatic recovery for the stall watchdog (issue #414, ADR-0103): the maximum number of in-place re-subscribes (Connection::resubscribe_consumer_in_place) the watchdog may drive per consumer, per stall streak.

At most one attempt is made per stall episode, and an episode can only close once per Self::consumer_stall_timeout window, so the bound is a cap on a sequence that is already rate-limited to one re-subscribe per window. The counter resets to zero on real progress only — one broker dispatch unit actually arriving (ConsumerState::record_dispatch_unit). It deliberately does NOT reset at the churn boundaries that zero the permit mirrors, because the recovery’s own re-subscribe is one of them and resetting there would make the bound infinite.

Default None — off, exactly like Self::consumer_stall_timeout. It is also inert without that knob: no stall window means no stall episode means nothing to recover from. Some(0) is the same as None (a budget of zero attempts).

A consumer the broker has reported as a Failover standby (ConsumerState::is_active == Some(false), issue #348) is skipped entirely: the stall is still reported, but no attempt is made and no budget is spent. A standby satisfies the stall predicate exactly as a wedged consumer does and never receives the dispatch unit that would reset the budget, so without that skip an armed recovery would spend its whole budget on every healthy standby in a failover group.

Each attempt zeroes this client’s permit mirrors and re-attaches this consumer id on the live socket, which repairs this client’s own slot in the broker’s dispatcher. Issue #414’s production failure was dispatcher-WIDE — the subscription’s availablePermits observed at -177300 across every attached consumer — and a fresh grant only lifts a corrupted aggregate by one receiver-queue window per attempt. That is precisely why the bound exists: once it is exhausted the client stops and logs the escalation (pulsar-admin topics unload) rather than re-subscribing forever against a fault it cannot repair. See docs/consumer-stall-recovery.md.

buggify: Buggify

Engine-arming slot for the ADR-0048 buggify fault-injection helper (ADR-0097 swarm configurations). Default crate::Buggify::disabled — a zero-sized no-op unless the buggify Cargo feature is enabled AND an engine chooses to honour the slot.

Only the moonpool engine reads this field: it installs the helper on every connection it constructs (initial dial and every supervised re-dial share the config, so the fire-counter map survives resets). The tokio engine deliberately ignores the slot, preserving ADR-0048’s guarantee that production binaries never see synthetic faults; buggify_off_is_nop.rs pins that contract.

Trait Implementations

impl Clone for ConnectionConfig

fn clone(&self) -> ConnectionConfig

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for ConnectionConfig

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

Formats the value using the given formatter.

impl Default for ConnectionConfig

fn default() -> ConnectionConfig

Returns the “default value” for a type.