You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/08/17 10:46:04 UTC
[PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
lhotari opened a new pull request, #26342:
URL: https://github.com/apache/pulsar/pull/26342
Related: #26340
### Motivation
The client memory limit is a producer's primary backpressure: it bounds the memory held by messages that have been queued but not yet acknowledged by the broker. #15723 added a safety net for clients that disable it, so that producers fall back to a bounded pending-message queue instead of buffering without any limit:
```java
public <T> ProducerBuilder<T> newProducer(Schema<T> schema) {
ProducerBuilderImpl<T> producerBuilder = new ProducerBuilderImpl<>(this, schema);
if (!memoryLimitController.isMemoryLimited()) {
producerBuilder.maxPendingMessages(NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES);
producerBuilder.maxPendingMessagesAcrossPartitions(
NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS);
}
return producerBuilder;
}
```
That net has two holes, and either one leaves a producer with no bound at all, because `ProducerImpl` only creates its semaphore `if (conf.getMaxPendingMessages() > 0)`:
1. It is only applied by the `newProducer(Schema)` overload. The no-argument `newProducer()` returns a plain builder, so it never gets the fallback. In-tree users of that overload include the Functions log appender and the WebSocket proxy's producer handler.
2. It is applied when the builder is constructed, so a later `maxPendingMessages(0)` overwrites it. Since `ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES` is 0, code that passes the default through explicitly silently disables the fallback rather than keeping it.
This came up while investigating #26340, where `pulsar-perf` exhausts direct memory against a slow broker. On branch-4.0 its producer hits both holes at once: it calls the no-argument `newProducer()` and then passes `maxPendingMessages(0)` through from the default. #26341 fixes that tool's own `--memory-limit` default; this PR closes the client-side holes that let the safety net be bypassed in the first place.
### Modifications
- Resolve the pending-message limits at producer creation, in `PulsarClientImpl`'s `createProducerAsync(conf, schema, interceptors)`, instead of on the builder. That is the funnel every producer built from this client passes through, so the fallback no longer depends on which `newProducer` overload created the builder, and cannot be undone by a later call setting a limit back to its unset value. Only unset limits are filled in; an explicit limit is never overwritten.
- Cap the per-producer fallback by the across-partitions limit. That limit is a budget shared by every partition, and its setter rejects a value below `maxPendingMessages`, so filling in the larger default first would throw `IllegalArgumentException` synchronously out of a method that returns a `CompletableFuture`.
- Resolve on a copy of the configuration, so filling in a limit does not leak into the next producer built from the same builder.
- Remove the now-redundant block from `newProducer(Schema)`. This also fixes a side effect it had: on a client with the memory limit disabled, `newProducer(schema).maxPendingMessagesAcrossPartitions(500)` used to throw, because the builder had already been given a `maxPendingMessages` of 1000.
- Document on `ProducerBuilder` what disabling either check actually means.
The V5 client reaches producer creation through `createSegmentProducerAsync` and is deliberately left unchanged here; it exposes no pending-message setting of its own and needs a separate decision. Note that `pulsar-perf` on master uses the V5 client, so this change on its own does not alter its behaviour there.
#### Noticed but deliberately left alone
When `maxPendingMessagesAcrossPartitions` is explicitly set below the topic's partition count, the per-partition share in `PartitionedProducerImpl` rounds down to 0, which means "no limit". So a tighter budget produces a looser bound, and an explicitly configured `maxPendingMessages` is silently discarded. On a client with the memory limit disabled it can also defeat the fallback applied here, since the filled-in limit is divided by that same code: an explicit budget of 500 on a topic with 501 partitions still ends up unbounded.
I tried clamping that share to at least 1 and reverted it. The branch it lives in is entered whenever the value is set, with no memory-limit condition, so the clamp also changes behaviour for memory-**limited** clients — the default configuration. There, an across-partitions budget smaller than the partition count currently yields no semaphore and a working, bytes-bounded producer; clamping turns it into a single permit, and with the default `blockIfQueueFull=false` the second concurrent in-flight message per partition fails with `ProducerQueueIsFullError`. That is a much larger blast radius than this PR should carry, so it needs its own change.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows. Each was confirmed to fail before the fix:
- `ProducerQueueSizeTest#testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled` — hole 1
- `ProducerQueueSizeTest#testLateZeroMaxPendingMessagesDoesNotDisableTheBoundWhenMemoryLimitDisabled` — hole 2
- `ProducerQueueSizeTest#testPartitionedProducerIsBoundedWhenMemoryLimitDisabled`
- `ProducerQueueSizeTest#testExplicitMaxPendingMessagesAboveTheFallbackDoesNotFailCreation` and `#testExplicitAcrossPartitionsLimitCapsTheFallback` — pin that filling in a default never fails producer creation
- `ProducerQueueSizeTest#testFallbackLeavesTheBuilderReusable` — pins that the resolved configuration is a copy
`ProducerQueueSizeTest#testMemoryLimitedClientKeepsUnboundedPendingMessages` pins that a client with a memory limit configured is unaffected.
Also run locally: `ProducerSemaphoreTest`, `ProducerMemoryLimitTest`, `MemoryLimitTest`, `ConsumerMemoryLimitTest`, `ProducerBuilderImplTest`, `PulsarClientImplTest`, `PartitionedProducerImplTest`, the `pulsar-websocket` module and `ProxyPublishConsumeTest`.
### Does this pull request potentially affect one of the following parts:
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [x] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
A producer created on a client whose memory limit is disabled now has a bounded pending-message queue where it previously had none. This is the behaviour #15723 intended; only the cases where it did not take effect change. Specifically:
- Because `maxPendingMessages` is a primitive `int` whose unset value is 0, an application that explicitly passed 0 to mean "unbounded" cannot be distinguished from one that never set it, and now gets the bound as well. Such an application can keep an unbounded message count by configuring a client memory limit, which bounds the queue by bytes instead, or by setting an explicit `maxPendingMessages`.
- On a partitioned topic, an application that set `maxPendingMessages` but left `maxPendingMessagesAcrossPartitions` unset now has the filled-in budget divided between the partitions, which can lower its per-partition limit. This matches what the `newProducer(Schema)` overload already did.
- The WebSocket proxy is affected out of the box, since `webSocketPulsarClientMemoryLimitInMB` defaults to 0 and its producers use the no-argument `newProducer()` with `blockIfQueueFull` false. A client with more than 1000 unacknowledged messages in flight now gets a failed `ProducerAck` instead of the proxy buffering them. Raising the limit through the `maxPendingMessages` query parameter currently also requires enabling batching, which is worth fixing separately.
### Documentation
- [ ] `doc-required`
- [x] `doc-not-needed`
- [ ] `doc`
- [ ] `doc-complete`
The behaviour change is documented in the `ProducerBuilder` javadoc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#issuecomment-5329323128
@lhotari LGTM, thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#issuecomment-5326125729
Now, the producer has backpressure, how to disable backpressure?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#issuecomment-5326826155
> Now, the producer has backpressure, how to disable backpressure?
good question. I'll check and revisit that. The `NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES`/`NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS ` solution should only get applied as a template for the producer builder. It might be necessary to expand the PR to cover the cli tools so that it's possible to detect that a parameter hasn't been explicitly set (using `Integer` instead of `int`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#issuecomment-5332708202
Thanks @nodece. One change since your review: I reverted the `pulsar-websocket` part, so this PR is now `pulsar-client` only.
The guard I had added there ignored a non-positive `maxPendingMessages` query parameter, but that treats the symptom. The proxy's actual problem is that `webSocketPulsarClientMemoryLimitInMB` is an `int` defaulting to 0, so the proxy always overrides the client's own 64M default with "no limit" — a memory limit is the right bound for it anyway, since it multiplexes many producers over one client. That is #26373, and the query parameter question belongs with it.
The PR description is updated accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#discussion_r3804755962
##########
pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java:
##########
@@ -508,7 +508,17 @@ private void popularProducerBuilderForServerSideEncrypt(ProducerBuilder<byte[]>
}
if (queryParams.containsKey("maxPendingMessages")) {
- builder.maxPendingMessages(Integer.parseInt(queryParams.get("maxPendingMessages")));
+ int maxPendingMessages = Integer.parseInt(queryParams.get("maxPendingMessages"));
+ if (maxPendingMessages > 0) {
Review Comment:
```suggestion
if (maxPendingMessages >= 0) {
```
Do you consider disabling backpressure by `maxPendingMessages`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#issuecomment-5328287568
> Now, the producer has backpressure, how to disable backpressure?
You were right — I've reworked the PR so that it can be disabled again.
**Why the first revision got this wrong.** #15723 introduced the 1000/50000 values to "restore ... pre-PIP-120 default configuration ... when [the memory] limit is disabled" (563a7cb2341), and PIP-120 (11bfc0e95dc) is the commit that both changed those defaults to 0 and documented 0 as "disable the pending messages check". So they are *defaults*, and an explicit 0 has to keep meaning "no message-count limit". The first revision turned them into a floor, because it inferred "unset" from `maxPendingMessages == 0` — which is the same value as the documented opt-out. With a primitive `int` there is nothing in `ProducerConfigurationData` that tells the two apart.
**What it does now.** `ProducerBuilderImpl` records whether the application actually called `maxPendingMessages(...)` / `maxPendingMessagesAcrossPartitions(...)`, or passed either to `loadConf`, and only a limit that was never configured gets a default filled in — at producer creation, on a copy of the configuration. On a client whose memory limit is disabled:
| configured | `maxPendingMessages` | `maxPendingMessagesAcrossPartitions` |
|---|---|---|
| nothing | 1000 | 50000 |
| `maxPendingMessages(500)` | 500 | 50000 |
| `maxPendingMessagesAcrossPartitions(500)` | 500 | 500 |
| `maxPendingMessagesAcrossPartitions(0)` | 1000 | 0 |
| `maxPendingMessages(0)` | 0 | 0 |
**So, to disable backpressure entirely — one call for each of the two mechanisms:**
```java
PulsarClient client = PulsarClient.builder()
.serviceUrl(url)
.memoryLimit(0, SizeUnit.BYTES) // no byte-based backpressure
.build();
Producer<byte[]> producer = client.newProducer()
.topic(topic)
.maxPendingMessages(0) // no message-count backpressure
.create();
```
An explicit `maxPendingMessages(0)` deliberately leaves the across-partitions budget alone as well, so that single call is enough on a partitioned topic too — filling the budget in would put a per-partition limit straight back, since `PartitionedProducerImpl` derives it from that budget. `ProducerQueueSizeTest#testExplicitZeroDisablesTheBoundWhenMemoryLimitDisabled` runs it for both topic shapes.
**Why the flag is on the builder and not on the configuration.** I tried the configuration first. `ProducerBuilder.loadConf` goes through `ConfigurationDataUtils.loadData`, which serialises the configuration to JSON and deserialises a *new* object through the public setters. A primitive is always emitted, so `setMaxPendingMessages(0)` always runs on the way back in: a `@JsonIgnore` marker would be set on every `loadConf` call, and a serialised one would be applied in `HashMap` key order. Changing the fields to `Integer` is worse — `getMaxPendingMessages()` would change return type with no possible bridge method, so anything compiled against the current jar gets a `NoSuchMethodError`, and the class is `Serializable`. The builder is the only place that sees what the application actually called.
**One bug found on the way, in its own commit.** `setMaxPendingMessagesAcrossPartitions` rejected a value below `maxPendingMessages`. Because `loadData` replays every property through the setters in an order the caller does not control, that makes `loadConf(Map.of("maxPendingMessages", 5000))` throw `maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages` on master today — `loadConf` cannot set a positive `maxPendingMessages` at all. The check is now `>= 0`; the relationship is enforced where it is used, in `PartitionedProducerImpl`. Happy to split that out if you would rather review it separately.
**The CLI tools.** I thought this PR would have to cover them, but on master they are all on the V5 client now: `--max-outstanding` / `-p` no longer reach a v4 producer at all. What *does* still bite there is `--memory-limit`: `PerformanceBaseArguments.memoryLimit` is a `long` with no initializer and `PerfClientUtils` passes it through unconditionally, so `pulsar-perf` disables the client memory limit unless the flag is given — which is exactly #26341. The `Integer`-instead-of-`int` change is needed for the branch-4.x backports of this PR, where `PerformanceProducer` calls the no-argument `newProducer()` and passes `maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES` (0) straight through; under these semantics that reads as an explicit "unbounded". I'll make those options `Integer` in the backport.
**V5** is deliberately untouched, and the reason is now in the code: it exposes no pending-message setting at all, so the client memory limit is both its only backpressure and the only thing an application can turn off. Filling in a message-count default there would leave a V5 user with no way to say "unbounded" at all.
**One thing added beyond the client:** the WebSocket proxy takes `maxPendingMessages` from a query parameter and its client has no memory limit by default, so now that an explicit 0 is honoured, `?batchingEnabled=true&maxPendingMessages=0` would let a remote client ask for an unbounded pending queue inside the shared proxy. A non-positive value is now ignored.
I've updated the PR description with the full behaviour table and the compatibility notes. Since the mechanism changed, could you take another look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #26342:
URL: https://github.com/apache/pulsar/pull/26342#discussion_r3804755962
##########
pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java:
##########
@@ -508,7 +508,17 @@ private void popularProducerBuilderForServerSideEncrypt(ProducerBuilder<byte[]>
}
if (queryParams.containsKey("maxPendingMessages")) {
- builder.maxPendingMessages(Integer.parseInt(queryParams.get("maxPendingMessages")));
+ int maxPendingMessages = Integer.parseInt(queryParams.get("maxPendingMessages"));
+ if (maxPendingMessages > 0) {
Review Comment:
```suggestion
if (maxPendingMessages >= 0) {
```
Do you consider disabling backpressure by `maxPendingMessages`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Apply no-memory-limit producer queue defaults at producer creation [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #26342:
URL: https://github.com/apache/pulsar/pull/26342
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org