# Client configuration and limitations

[Documentation](README.md) | [简体中文](client-guide.zh-CN.md)

Configuration semantics and interoperability caveats. For verification scope, see the [capability matrix](capabilities.md).

## Features

- Connection management over TCP with the Pulsar binary protocol (`pulsar://`), handshake, keepalive and request/response correlation
- `ClientOptions` for connection/operation timeouts, keepalive interval, per-broker connection pool size and idle eviction, a shared byte budget for producer messages, chunks, receive queues and inbound frames, advertised listener and lookup properties; asynchronous authentication providers and token-file reload on connection or challenge
- Topic lookup with redirect following and connection pooling
- Producer: synchronous and asynchronous send, shared/exclusive/wait-for-exclusive/fencing access modes, producer metadata, batching (count/bytes/delay triggers, manual flush), broker receipts, send error propagation, optional message-count and client-wide payload-byte limits with blocking or fail-fast admission, and send timeout including buffered batches
- Producer interceptors can transform a message before partition routing and observe either its broker receipt or send failure, including asynchronous sends that the caller never waits for.
- Consumer interceptors observe application delivery, `receive()` / `try_receive()` failures, and the results of ACK and negative ACK operations, including `Message::ack`, `Message::nack`, and ACK by ID.
- `ClientOptions::new(on_event=..., on_span=...)` exposes structured events and paired send, receive, ACK and reconnect spans. `ClientMetrics` provides outcome counters, duration histograms and Prometheus text export.
- Consumer: Exclusive / Shared / Failover / KeyShared subscriptions (including auto-split and sticky hash-range policies), individual, cumulative and ID-list ACK with optional broker confirmation, negative ack with optional exponential redelivery backoff, ack timeout, seek by message ID or timestamp, unsubscribe, flow control (FLOW permits), non-partitioned single-topic zero-sized receive queue with on-demand FLOW, optional auto-scaled receive queue, batched message unpacking and optional per-index batch ACK
- Dead-letter policy: route messages after the configured number of unsuccessful deliveries, preserving payload, key, ordering key, properties, and event time; acknowledge the source only after the dead-letter producer receives a broker receipt
- Retry-letter consumers: use `retry_topic` on ordinary subscriptions or the explicit constructor, then `reconsume_later` to publish with a delay and route exhausted retries to the dead-letter topic
- Schema registry lookup by latest or specific version; producers attach the broker-assigned version and messages expose it through `schema_version()`

`DeadLetterPolicy::new(1U, dead_letter_topic)` allows one application attempt;
after a NACK, the next delivery goes to the dead-letter topic. The policy can
be passed to single-topic, multi-topic, and pattern consumer constructors.
- Explicit multi-topic consumers and pattern consumers that discover added topics and close removed sources in a namespace. Broker-initiated source closure is reconciled by the next discovery refresh, so deleting an active topic does not immediately recreate it.
- Bounded merge queues for composite consumers and readers, so slow applications stop source forwarding
- Automatic reconnection: producers replay unconfirmed messages and consumers re-subscribe after a broker connection breaks; `ClientOptions` can set backoff and a finite retry count
- `max_memory_bytes` charges producer messages (payload, keys and properties), encoded replay frames, incomplete chunks, queued consumer/reader messages (payload and dynamic metadata), and inbound broker frames from their declared length through dispatch. Producer reservations release on receipt, error, timeout or close; chunk reservations release on completion, eviction or clear; receive reservations release on application delivery, seek or close. The encoded frame must fit alongside its source message; frame admission fails with `ClientMemoryFull` if it cannot fit. A receive reservation that exceeds the remaining shared budget closes its consumer/reader queue with `ClientMemoryFull`; an oversized inbound frame closes its broker connection and reports `ClientMemoryFull` to affected consumers and readers without retrying that frame. A new client with a sufficient budget can replay it. TLS/socket internal buffers and runtime heap overhead are not counted.
- Binary-protocol authentication: pluggable `Authentication` trait, token and basic auth built in; rotating token files, OAuth2 client credentials, externally supplied Athenz role tokens, and `athenz_zts_provider(AthenzZtsOptions::new(...))` for RSA service NToken or client-certificate exchange with ZTS. Pass any provider through `ClientOptions::new(auth_provider=...)`; broker challenges refresh the supplied auth data
- TLS via `pulsar+ssl://` service URLs, with system roots or a custom PEM CA through `Client::connect_tls_with_ca`. `ClientOptions` also accepts `TlsIdentity::new(cert_file, key_file)` or an async `tls_identity_provider`, TLS 1.2/1.3 bounds, an OpenSSL TLS 1.2 cipher list, and TLS 1.3 ciphersuites. The native adapter loads a fresh identity for each connection and preserves CA and hostname checks across lookups and reconnects.
- Payload compression: LZ4 (frame format), Zlib, Zstd and raw Snappy; see the Snappy interoperability limitation below
- Partitioned topics: configurable key hash (default), round-robin, fixed single-partition, or custom callback routing; producers expose the last confirmed sequence ID; consumers aggregate all partitions and both producers and consumers discover added partitions (60-second default polling interval)
- Optional explicit or broker-limit-driven automatic producer chunking and consumer/reader chunk reassembly, including compressed payloads; `ChunkAssemblyPolicy` limits pending assemblies and expires incomplete chunks even when no further messages arrive
- Reader API: non-durable replay from any message id with optional inclusive start/seek, `seek`, `has_message_available`, last-message ID, custom name/properties/subscription, aggregation and added-partition discovery
- TableView: raw-byte or schema-decoded key/value snapshot from a compacted reader, live updates, tombstone deletion, and change listeners; typed listeners receive decode failures as `Err`
- Transactions: coordinator ownership lookup, `new_transaction` / `commit` / `abort`, transactional send and ack
- Admin REST API: topic create/delete and properties, partitioned topic create/expand/delete and metadata, list topics and partitioned topics, create/list/delete subscriptions, skip or expire subscription backlog, topic and partitioned-topic stats
- Schema declaration: producers/consumers declare `SchemaInfo` on creation; `SchemaCodec[T]` encodes and decodes STRING, JSON, BYTES, numeric primitives, Avro, Protobuf Native and legacy Protobuf generated messages, or uses custom callbacks
- Message encryption: `MessageCrypto` loads RSA or EC PEM keys, wraps a fresh AES-256-GCM key with RSA-OAEP-SHA1 or Pulsar Java-compatible ECIES for each named recipient, and decrypts in consumers, readers and TableViews. Pass `message_crypto` and `encryption_key_names` to encrypted producers; encrypted retry/DLQ consumers need both options to re-encrypt forwarded messages. Consumers and readers accept `decryption_failure_action=Fail | Consume | Discard`.
- Delayed delivery: `deliver_at` / `deliver_after` on `ProducerMessage`
- Timestamp seek for readers

## Configuration notes and known limitations

- Pulsar Snappy messages use raw blocks, matching the pinned Go client. The public `snappy_compress` and `snappy_decompress` helpers still handle Google framed streams. Java clients use xerial framing and cannot decode raw Snappy; the official Java and Go clients have the same incompatibility
- The tested Pulsar 4.2.4 broker rejected creation of partitioned metadata over an existing non-partitioned topic with HTTP 409. Migrate data into a new partitioned topic when this change is needed.
- Retry-letter handling is available through `retry_topic` on single, multi-topic and pattern consumers, or the explicit `create_retry_consumer` constructor. Athenz ZTS service-key exchange signs a fresh RSA NToken; certificate mode uses `AthenzZtsOptions::new(..., certificate_file=..., ca_pem_file=...)` with an HTTPS ZTS URL and the certificate's private key. Both modes fetch a role token, retry transient failures and refresh before expiry; the ZTS URL is the server base URL before `/zts/v1`. OAuth2 currently requests a fresh token for each connection or challenge. External OAuth2 identity providers and external Athenz services remain unverified.
- `MessageCrypto::load_public_key_file` and `load_private_key_file` can be called again after replacing PEM files; new sends and reads use the replaced key. Decryption defaults to `Fail`, which reports `InvalidFrame` and stops the affected consumer/reader. `Consume` delivers the raw ciphertext with `Message::decryption_failed() == true` so the application can decide whether to ACK; encrypted batches cannot be delivered this way. `Discard` skips the message and sends an individual ACK from a consumer; a reader skips it without an ACK. RSA-OAEP-SHA1 and ECIES wrap AES-256-GCM keys; ECIES uses the native OpenSSL 3 adapter. The native random source is `/dev/urandom`. The pinned Go client's default crypto accepts only RSA keys.
- `MessageCrypto::new(public_key_reader=..., private_key_reader=...)` accepts synchronous callbacks from key name to PEM bytes. Readers are queried for each send or encrypted delivery and override keys added directly; returning `None` reports an unavailable key.
- `SchemaCodec::avro(definition, to_datum, from_datum)` parses an Avro schema and converts typed values to binary Avro datums; `avro_datum(definition)` exposes the full Avro datum model directly. Add `yugonlian/moon-avro@0.3.0` and import its `codec` package to construct or match datums. `Client::decode_avro(message, reader_codec)` looks up the message's writer schema version, caches it per topic/version, resolves field aliases and defaults, unions, collections, enum defaults and allowed numeric promotions, then returns the reader codec's type. It reports `SchemaIncompatible` for incompatible schemas. The synchronous codec still decodes with its declared schema.
- `SchemaCodec::protobuf_native(descriptor_set, root_file, root_message)` accepts a binary `FileDescriptorSet` and generated MoonBit Protobuf message type. The root must exist in the descriptor set. `SchemaCodec::protobuf(definition)` uses the legacy PROTOBUF schema with an Avro-style JSON definition and generated MoonBit Protobuf message type. INT16/32/64 codecs default to the Java client's big-endian encoding; pass `little_endian=true` for the Go client's integer encoding.
- Automatic chunking requires the broker to advertise a maximum message size. `ChunkAssemblyPolicy::new(auto_ack_incomplete=true)` makes consumers and readers acknowledge received chunks when an incomplete assembly is evicted or expires; the default leaves them unacknowledged for broker redelivery.
- Batch messages use whole-entry ACK after every index is acknowledged by default. Set `enable_batch_index_ack=true` on a consumer to acknowledge each index; the broker must enable `acknowledgmentAtBatchIndexLevelEnabled`. Partial batch ACK cannot request broker confirmation in the default mode. Transactional ACK supports both modes and waits for each broker ACK response; the broker must have transactions enabled.
- The pinned Go client's consumer exposes the 16-byte AES-GCM tag for an encrypted empty payload sent by MoonBit because it skips replacing the payload buffer when `UncompressedSize` is zero. MoonBit correctly decrypts Go's encrypted empty payload; `interop/go/encryption receive-empty` reproduces the Go-side limitation.
- Set `ack_grouping=AckGroupingOptions::new(max_size=1000, max_time_ms=100)` on a consumer to buffer ACKs by count or time. Grouping is off unless configured; confirmed and transactional ACKs flush the buffer and send immediately. Closing or seeking flushes buffered ACKs, while reconnecting drops them so the broker can redeliver.
- `Consumer::ack_ids([id1, id2])` combines ordinary IDs into one ACK frame; batch-index IDs retain their bitmap handling. `Consumer::last_message_ids()` and `Reader::last_message_ids()` return a map keyed by source Topic, including partition Topics. A zero receive queue is accepted only for a non-partitioned single-topic consumer; partitioned, multi-topic, retry and pattern subscriptions reject it.
- `send_timeout_ms` starts after message-count and memory admission, before encoding or batch buffering. One deadline covers batching, all chunks and reconnect/replay; flush and retry do not reset it. Time waiting for admission is excluded and can be bounded by caller cancellation. Expiry removes buffered messages and stops replay once no live message remains in a frame. Each producer uses one deadline scheduler and immediately removes completed messages from its indexed deadline heap. A timeout after SEND does not prove the broker rejected the message, so applications should use stable producer names and sequence IDs when retrying.
- `ClientOptions::new(max_memory_bytes=...)` shares a retained-byte budget across producers, incomplete chunks, queued received messages and inbound broker frames. A producer with `block_if_queue_full=false` reports `ClientMemoryFull` when the budget is full; blocking producers wait for a receipt, failure, timeout, delivery or client close to release it. This is a byte count of protocol and message data, not a precise process heap limit.
- Set `auto_scaled_receiver_queue=true` on a consumer to start with one FLOW permit and double the prefetch size after a full queue is followed by an empty read, up to `receiver_queue_size`. A zero maximum cannot be combined with auto scaling. Composite consumers pass this option to each source; their merge queue still has its own buffering.
- Pass `interceptors=[ProducerInterceptor::new(before_send=..., on_send_result=...)]` to `Client::create_producer`. Hooks run in order on the logical producer; `on_send_result` also receives immediate admission errors such as `ProducerQueueFull`. Callbacks are synchronous and cannot raise errors.
- Pass `interceptors=[ConsumerInterceptor::new(before_consume=..., on_receive_error=..., on_ack=..., on_nack=...)]` to a consumer constructor. Hooks run on application delivery, a failed `receive()` / `try_receive()`, and when its message or ID ACK/nack call completes. `on_ack` reports a rejected confirmed ACK as an error; an unconfirmed ACK callback means the frame was submitted, not broker-confirmed.
- Use `let metrics = ClientMetrics::new()` and `ClientOptions::new(url, on_event=fn(event) { metrics.record(event) }, on_span=fn(span) { metrics.record_span(span) })` for counters and duration histograms; `metrics.prometheus_text()` exports bounded-label Prometheus metrics. `ClientSpanEvent::Started` and `Finished` share a correlation ID, operation, Topic and optional message ID; the finish event includes duration, outcome and error text. Callbacks run synchronously. `ClientMetrics::snapshot` continues to expose outcome counters.
- `Connection::max_message_size` exposes the negotiated broker limit; producers cap encoded frames to that limit and can select a chunk size from it automatically.
- Ordinary `pulsar+ssl://` connections use `moonbitlang/async/tls`; identity and protocol/cipher options select the repository's OpenSSL 3 native adapter. Client certificates automatically announce Pulsar's `tls` authentication method unless another auth provider is configured. Windows remains unverified; see the capability matrix for recorded platform and Broker verification.