[Enhancement] Configurable memory limit for broker, proxy and WebSocket proxy Pulsar clients, shared across all of them

[lhotari]

Search before asking

• I searched the issues and found nothing similar.

Motivation

Every Pulsar client created inside a broker explicitly disables the client memory limit, so there is no bound — per client or in aggregate — on the direct memory those clients can hold in their producer pending queues and consumer receive queues:

Site	Code
pulsar-broker/.../broker/PulsarService.java:1992	initialConf.setMemoryLimitBytes(0); — comment: "Disable memory limit for broker client and disable stats"
pulsar-broker/.../broker/service/BrokerService.java:1664	clientBuilder.memoryLimit(0, SizeUnit.BYTES); in getReplicationClient(...) — cached one per remote cluster
pulsar-broker/.../broker/namespace/NamespaceService.java:1753	.memoryLimit(0, SizeUnit.BYTES) in getNamespaceClient(...) — cached one per remote cluster
pulsar-broker/.../compaction/CompactorTool.java:82	.memoryLimit(0, SizeUnit.BYTES)

A broker geo-replicating to N remote clusters therefore holds 1 + 2N unlimited PulsarClient instances. Each one independently buffers, and nothing caps the total. ServiceConfiguration has no brokerClientMemoryLimit* key at all — the only memory-limit setting in it is webSocketPulsarClientMemoryLimitInMB (ServiceConfiguration.java:3776), which also defaults to 0 (disabled).

The same shape applies to the Pulsar proxy (ProxyConfiguration has no memory-limit key either) and to the WebSocket proxy (has one, but disabled by default and expressed only in whole MB).

The client-side machinery to fix this already exists and is unused here. PIP-234 (#19074) added PulsarClientSharedResources, and #25477 added a shared MemoryLimitController to it (SharedResource.MemoryLimitController, PulsarClientSharedResourcesBuilder.configureMemoryLimitController(...)). The broker does not use PulsarClientSharedResources at all — rg PulsarClientSharedResources pulsar-broker/src/main/ returns no matches. It predates the API and hand-rolls sharing, passing ioEventLoopGroup, brokerClientSharedTimer and four shared executor providers individually into PulsarClientImpl.builder() in PulsarService.createClientImpl, and never a memoryLimitController.

Both cluster-scoped client factories funnel through createClientImpl, so a single shared controller can cover all 1 + 2N clients.

Proposal

Add a memory-limit configuration for server-side Pulsar clients, applied as a single shared budget across all clients a process creates.

1. Broker — a new ServiceConfiguration key, e.g. brokerClientMemoryLimit, backing a shared MemoryLimitController installed in PulsarService.createClientImpl, so the broker client, all replication clients and all namespace clients draw on one budget. When only one client exists the same key degenerates naturally to a plain per-client memory limit — no separate setting is needed for that case.

2. Pulsar proxy — the equivalent key in ProxyConfiguration, shared across the proxy's client instances.

3. WebSocket proxy — the existing webSocketPulsarClientMemoryLimitInMB should gain the same treatment. Note the WebSocket proxy also creates producers with blockIfQueueFull(false), so a limit here changes failure behaviour rather than adding blocking; see #26343.

4. Value format. Accept both a plain byte count and a unit-suffixed string — "64M", "64m", "1G", "512K" — rather than the ...InMB integer convention. A parser already exists: org.apache.pulsar.cli.converters.ByteUnitUtil.validateSizeString, used by ByteUnitToLongConverter in pulsar-cli-utils. Broker/proxy config fields go through FieldParser in pulsar-common, whose stringToLong (FieldParser.java:285) has no unit support today, so this needs either a shared parser moved/added to pulsar-common or a dedicated config type. The v5 client API already models this as MemorySize (pulsar-client-api-v5/.../config/MemorySize.java), which is worth aligning with.

5. Default. Deliberately open for discussion in the PIP. Keeping 0 (disabled) preserves today's behaviour; any non-zero default is a behaviour change for existing deployments, since replication and other internal clients would start applying backpressure where they previously buffered without limit.

Prerequisite

The shared-controller defects in #26345 should be fixed first — in particular, the shared limit currently defaults to 0/unlimited and silently overrides each client's own memoryLimit(...). Wiring the broker onto a shared controller while that controller defaults to unlimited would achieve nothing.

Existing partial escape hatch

Per-client limits are already reachable today, undocumented: brokerClient_memoryLimitBytes overrides the hard-coded 0, because every setMemoryLimitBytes(0) / .memoryLimit(0, ...) call runs before the corresponding loadConf(filterAndMapProperties(..., "brokerClient_")). So a per-cluster limit is settable; the aggregate bound across clients is what is missing, along with any documented, first-class configuration.

Scope & compatibility

• Adds new broker and proxy configuration keys, and a new value format for them → this requires a new PIP. It builds directly on PIP-234 (PIP-234: Support using shared thread pool across multiple Pulsar client instance  #19074) and the shared memory limit controller added in [improve][client] PIP-234: Support sharing the memory limit controller across multiple isolated Pulsar client instances #25477.
• If the default stays 0, there is no behaviour change for existing deployments; any other default must be called out in release notes, since internal clients would begin applying backpressure.
• No wire-protocol, metadata-format or client-server compatibility impact.
• Consider whether the shared budget should be split (e.g. replication vs. other internal clients) or remain a single pool — a single pool is simpler but lets one busy remote cluster starve the rest. Worth resolving in the PIP.

Related

• PIP-234 / PIP-234: Support using shared thread pool across multiple Pulsar client instance  #19074 — shared client resources umbrella.
• [Enhancement] PIP-234: Add a solution to share the memory limit controller solution #25212 (closed, completed) / [improve][client] PIP-234: Support sharing the memory limit controller across multiple isolated Pulsar client instances #25477 (merged) — the shared MemoryLimitController this builds on.
• [Bug] Shared MemoryLimitController silently disables the client memory limit and drops the client memory metrics #26345 — shared MemoryLimitController silently disables the limit and drops metrics (prerequisite for this).
• [Bug] Producer send admission control blocks Pulsar's own internal threads, and there is no non-blocking backpressure mode #26343 — producer send admission control blocks the calling thread; relevant to what "hitting the limit" does on the broker's own threads once a limit actually exists.
• [Bug] Pulsar-perf producer OutOfDirectMemoryError with slow broker #26340 / [fix][cli] Restore the 64M default client memory limit in pulsar-perf #26341 / [fix][client] Apply no-memory-limit producer queue defaults at producer creation #26342 — the client memory-limit default work that surfaced this area.