apache / apache/pulsar

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

Open
#26,346 0 comments 0 reactions 1 assignee Claimed by @lhotari View on GitHub
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before asking

- [x] I searched the [issues](https://github.com/apache/pulsar/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 (#19074) and the shared memory limit controller added in #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 / #19074 — shared client resources umbrella.
- #25212 (closed, completed) / #25477 (merged) — the shared `MemoryLimitController` this builds on.
- #26345 — shared `MemoryLimitController` silently disables the limit and drops metrics (prerequisite for this).
- #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.
- #26340 / #26341 / #26342 — the client memory-limit default work that surfaced this area.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.