From 98e0519ca41ac6b3f34de50e9128b4409df3c65a Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Mon, 17 Aug 2026 13:00:53 +0300 Subject: [PATCH 1/4] [fix][client] Apply no-memory-limit producer queue defaults at producer creation ### 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 ProducerBuilder newProducer(Schema schema) { ProducerBuilderImpl 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. ### 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. ### Verifying this change Added tests, each confirmed to fail before the fix: - `ProducerQueueSizeTest#testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled` (hole 1) - `ProducerQueueSizeTest#testLateZeroMaxPendingMessagesDoesNotDisableTheBoundWhenMemoryLimitDisabled` (hole 2) - `ProducerQueueSizeTest#testPartitionedProducerIsBoundedWhenMemoryLimitDisabled` - `ProducerQueueSizeTest#testExplicitMaxPendingMessagesAboveTheFallbackDoesNotFailCreation` and `#testExplicitAcrossPartitionsLimitCapsTheFallback`, which pin that filling in a default never fails producer creation - `ProducerQueueSizeTest#testFallbackLeavesTheBuilderReusable`, which pins that the resolved configuration is a copy `ProducerQueueSizeTest#testMemoryLimitedClientKeepsUnboundedPendingMessages` pins that a client with a memory limit configured is unaffected. Noticed while working on this, left alone as a separate concern: 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. It affects clients regardless of their memory limit, and clamping the share turned out to change behaviour for memory-limited clients too, so it needs its own change rather than riding along here. ### Does this pull request potentially affect one of the following parts: - [x] The default values of configurations 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. --- .../client/api/ProducerQueueSizeTest.java | 194 ++++++++++++++++++ .../pulsar/client/api/ProducerBuilder.java | 11 +- .../pulsar/client/impl/PulsarClientImpl.java | 67 ++++-- 3 files changed, 258 insertions(+), 14 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java index ba28447eae804..11e054ac99f0a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java @@ -18,16 +18,31 @@ */ package org.apache.pulsar.client.api; +import static org.assertj.core.api.Assertions.assertThat; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; import lombok.Cleanup; import org.apache.pulsar.broker.service.SharedPulsarBaseTest; +import org.apache.pulsar.client.impl.ProducerBase; +import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; public class ProducerQueueSizeTest extends SharedPulsarBaseTest { + /** + * The bounds {@code PulsarClientImpl} falls back to when the client memory limit is disabled. + * Duplicated here on purpose: these are a documented client default, so a change to them should + * break a test rather than pass silently. + */ + private static final int NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES = 1000; + private static final int NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS = 50000; + + private static ProducerConfigurationData confOf(Producer producer) { + return ((ProducerBase) producer).getConfiguration(); + } + @DataProvider(name = "matrix") public Object[][] matrix() { return new Object[][]{ @@ -72,4 +87,183 @@ public void testRemoveMaxQueueLimit(boolean blockIfQueueFull, boolean partitione f.get(); } } + + /** + * A client with the memory limit disabled has no byte-based backpressure, so producers must fall + * back to a bounded pending-message queue. This has to hold for the no-argument + * {@code newProducer()} overload as well, not just {@code newProducer(Schema)}. + */ + @Test + public void testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer().topic(newTopicName()).create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + + /** + * Setting the pending-message limits to 0 after the builder has been created must not leave a + * producer with no bound at all when the client memory limit is disabled: 0 means "unset", and + * unset falls back to the bounded defaults. + */ + @SuppressWarnings("deprecation") + @Test + public void testLateZeroMaxPendingMessagesDoesNotDisableTheBoundWhenMemoryLimitDisabled() + throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer(Schema.BYTES) + .topic(newTopicName()) + .maxPendingMessages(0) + .maxPendingMessagesAcrossPartitions(0) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + + /** + * {@code maxPendingMessagesAcrossPartitions} must be {@code >= maxPendingMessages}. Filling in + * the across-partitions fallback must therefore never lower it below an explicitly configured + * per-partition limit, which would fail producer creation. + */ + @Test + public void testExplicitMaxPendingMessagesAboveTheFallbackDoesNotFailCreation() throws Exception { + int maxPendingMessages = NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS + 10_000; + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .maxPendingMessages(maxPendingMessages) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(maxPendingMessages); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isGreaterThanOrEqualTo(maxPendingMessages); + } + + /** + * Filling in the fallback must not write it back into the builder's own configuration. The + * builder stays reusable, and a limit set on it afterwards is still validated against what the + * caller configured rather than against a filled-in default. + */ + @SuppressWarnings("deprecation") + @Test + public void testFallbackLeavesTheBuilderReusable() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + ProducerBuilder builder = client.newProducer(); + + @Cleanup + Producer first = builder.topic(newTopicName()).create(); + assertThat(confOf(first).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + + // Rejected if creating the first producer had left the fallback in the builder, since the + // across-partitions limit has to be >= maxPendingMessages. + @Cleanup + Producer second = builder.topic(newTopicName()) + .maxPendingMessagesAcrossPartitions(500) + .create(); + assertThat(confOf(second).getMaxPendingMessages()).isEqualTo(500); + } + + /** + * The fallback has to reach partitioned producers too, where the per-partition queue is derived + * from the across-partitions budget. + */ + @Test + public void testPartitionedProducerIsBoundedWhenMemoryLimitDisabled() throws Exception { + String topic = newTopicName(); + admin.topics().createPartitionedTopic(topic, 10); + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer().topic(topic).create(); + + // The budget spread over 10 partitions is well above the per-producer default, so each + // partition keeps the full default. + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + + /** + * The across-partitions limit is a budget shared by every partition, so the per-producer + * fallback must be capped by it. Otherwise the fallback would exceed an explicitly configured + * budget, which producer creation rejects. + */ + @SuppressWarnings("deprecation") + @Test + public void testExplicitAcrossPartitionsLimitCapsTheFallback() throws Exception { + int maxPendingMessagesAcrossPartitions = 500; + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .maxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(maxPendingMessagesAcrossPartitions); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) + .isEqualTo(maxPendingMessagesAcrossPartitions); + } + + /** + * The fallback only exists to replace the missing byte-based backpressure. When a memory limit + * is configured, an unset pending-message limit keeps meaning "no message-count limit". + */ + @Test + public void testMemoryLimitedClientKeepsUnboundedPendingMessages() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(64, SizeUnit.MEGA_BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer().topic(newTopicName()).create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } } diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java index 7b35432da1d6a..1cb6db79e9313 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java @@ -174,7 +174,11 @@ public interface ProducerBuilder extends Cloneable { * the client application. Until the producer gets a successful acknowledgment back from the broker, * it will keep in memory (direct memory pool) all the messages in the pending queue. * - *

Default is 0, which disables the pending messages check. + *

Default is 0, which disables the pending messages check. Disabling it only removes the + * message-count limit; the memory the pending queue may hold is then bounded by the client + * memory limit ({@link ClientBuilder#memoryLimit(long, SizeUnit)}) instead. When the client + * memory limit is also disabled there would be no backpressure left at all, so producers fall + * back to a default pending messages queue size rather than buffering without limit. * * @param maxPendingMessages * the max size of the pending messages queue for the producer @@ -190,7 +194,10 @@ public interface ProducerBuilder extends Cloneable { * The purpose of this setting is to have an upper-limit on the number * of pending messages when publishing on a partitioned topic. * - *

Default is 0, which disables the pending messages across partitions check. + *

Default is 0, which disables the pending messages across partitions check. As with + * {@link #maxPendingMessages(int)}, a producer created on a client whose memory limit is + * disabled falls back to a default budget instead, since no backpressure would otherwise be + * left. * *

If publishing at a high rate over a topic with many partitions (especially when publishing messages without a * partitioning key), it might be beneficial to increase this parameter to allow for more pipelining within the diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index 873069c1c930f..f30ba14fa4cd2 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -609,17 +609,9 @@ public ProducerBuilder newProducer() { return new ProducerBuilderImpl<>(this, Schema.BYTES); } - @SuppressWarnings("deprecation") @Override public ProducerBuilder newProducer(Schema schema) { - ProducerBuilderImpl producerBuilder = new ProducerBuilderImpl<>(this, schema); - if (!memoryLimitController.isMemoryLimited()) { - // set default limits for producers when memory limit controller is disabled - producerBuilder.maxPendingMessages(NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES); - producerBuilder.maxPendingMessagesAcrossPartitions( - NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); - } - return producerBuilder; + return new ProducerBuilderImpl<>(this, schema); } @Override @@ -701,19 +693,70 @@ public CompletableFuture> createProducerAsync(ProducerConfigurat + " Topic: '" + topic + "'")); } + final ProducerConfigurationData producerConf = resolvePendingMessagesLimits(conf); + if (schema instanceof AutoProduceBytesSchema) { AutoProduceBytesSchema autoProduceBytesSchema = (AutoProduceBytesSchema) schema; if (autoProduceBytesSchema.hasUserProvidedSchema()) { - return createProducerAsync(topic, conf, schema, interceptors); + return createProducerAsync(topic, producerConf, schema, interceptors); } return reloadSchemaForAutoProduceProducer(topic, autoProduceBytesSchema) - .thenCompose(schemaInfoOptional -> createProducerAsync(topic, conf, schema, interceptors)); + .thenCompose(schemaInfoOptional -> + createProducerAsync(topic, producerConf, schema, interceptors)); } else { - return createProducerAsync(topic, conf, schema, interceptors); + return createProducerAsync(topic, producerConf, schema, interceptors); } } + /** + * Resolve the producer's pending-message limits before the producer is created. + * + *

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. When it is disabled + * there is nothing left to bound that queue, so fall back to message-count limits rather than + * let a producer that outruns its broker buffer without any limit at all. + * + *

Only limits that are unset are filled in; an explicit limit is never overwritten. Note that + * on a partitioned topic a filled-in across-partitions budget is still divided between the + * partitions afterwards, which can lower an explicitly configured per-producer limit. Doing this + * at creation time rather than on the builder means the fallback cannot be missed depending on + * which {@code newProducer} overload produced the builder, nor undone by a later call setting a + * limit back to its unset value. Segment producers created for the V5 client go through + * {@link #createSegmentProducerAsync} and do not pass through here. + * + * @param conf the requested producer configuration + * @return the configuration to create the producer with; a resolved copy when a fallback + * applies, otherwise {@code conf} unchanged + */ + private ProducerConfigurationData resolvePendingMessagesLimits(ProducerConfigurationData conf) { + if (memoryLimitController.isMemoryLimited() + || (conf.getMaxPendingMessages() > 0 && conf.getMaxPendingMessagesAcrossPartitions() > 0)) { + return conf; + } + int maxPendingMessages = conf.getMaxPendingMessages() > 0 + ? conf.getMaxPendingMessages() + : NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES; + int maxPendingMessagesAcrossPartitions = conf.getMaxPendingMessagesAcrossPartitions() > 0 + ? conf.getMaxPendingMessagesAcrossPartitions() + : Math.max(maxPendingMessages, NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + // The across-partitions limit is a budget shared by every partition, so a single producer's + // queue can never exceed it. This also keeps the two limits consistent for the setters below. + maxPendingMessages = Math.min(maxPendingMessages, maxPendingMessagesAcrossPartitions); + + // Resolve on a copy: the builder hands over its own configuration instance, so filling in a + // limit here would otherwise leak into the next producer built from the same builder. + ProducerConfigurationData resolved = conf.clone(); + // Order matters: the across-partitions setter rejects a value below maxPendingMessages. + resolved.setMaxPendingMessages(maxPendingMessages); + resolved.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); + log.debug().attr("topic", conf.getTopicName()) + .attr("maxPendingMessages", maxPendingMessages) + .attr("maxPendingMessagesAcrossPartitions", maxPendingMessagesAcrossPartitions) + .log("Client memory limit is disabled, applying default producer pending message limits"); + return resolved; + } + @SuppressWarnings("unchecked") public CompletableFuture reloadSchemaForAutoProduceProducer(String topic, AutoProduceBytesSchema autoSchema) { return lookup.getSchema(TopicName.get(topic)).thenAccept(schemaInfoOptional -> { From e535ba765ba8dbdd0bc9bd90598a100f9c403f37 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 18 Aug 2026 15:37:40 +0300 Subject: [PATCH 2/4] [fix][client] Allow maxPendingMessagesAcrossPartitions to be lower than maxPendingMessages ### Motivation `ProducerConfigurationData.setMaxPendingMessagesAcrossPartitions` rejected any value below `maxPendingMessages`. That makes the two setters order-dependent, and it makes `ProducerBuilder.loadConf` fail outright for any positive `maxPendingMessages`: `ConfigurationDataUtils.loadData` serialises the configuration, merges the caller's map and deserialises a new instance by replaying every property through the public setters, in an order the caller does not control. So builder.loadConf(Map.of("maxPendingMessages", 5000)) throws `maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages`, because the across-partitions property that comes along with the merged map is still at its default of 0. The same check also makes a builder reject a legitimate call sequence: setting a per-producer limit and then a smaller shared budget throws, while the reverse order is accepted. ### Modifications Validate only that the value is not negative. The relationship between the two limits is enforced where it is used: `PartitionedProducerImpl` lowers the per-partition limit to its share of the budget whenever a budget is set, and the budget is meaningless on a non-partitioned topic. Assisted-by: Claude Code (Opus 5) --- .../impl/conf/ProducerConfigurationData.java | 12 +++++-- .../client/impl/ProducerBuilderImplTest.java | 32 ++++++++++++++++++- 2 files changed, 41 insertions(+), 3 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java index 601cf78c8b893..02c25bece9053 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java @@ -252,9 +252,17 @@ public void setMaxPendingMessages(int maxPendingMessages) { this.maxPendingMessages = maxPendingMessages; } + /** + * The across-partitions budget used to be rejected when it was below {@link #maxPendingMessages}, + * which made the two setters order-dependent: it depended on which of them had been called first, + * and it made {@code loadConf} fail outright for any positive {@code maxPendingMessages}, since + * that replays every property through the setters in an order the caller does not control. The + * relationship is enforced where it is used instead — {@code PartitionedProducerImpl} lowers the + * per-partition limit to the share of the budget when a budget is set. + */ public void setMaxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) { - checkArgument(maxPendingMessagesAcrossPartitions >= maxPendingMessages, - "maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages"); + checkArgument(maxPendingMessagesAcrossPartitions >= 0, + "maxPendingMessagesAcrossPartitions needs to be >= 0"); this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions; } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java index 7554135943194..86468b4badb97 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java @@ -23,6 +23,7 @@ import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @@ -119,6 +120,21 @@ public void testProducerBuilderImplWhenMessageRoutingModeIsRoundRobinPartition() assertNotNull(producer); } + /** + * {@code loadConf} rebuilds the configuration by replaying every property through the public + * setters, and {@code setMaxPendingMessagesAcrossPartitions} rejects a value below + * {@code maxPendingMessages}. Pins that loading a positive limit does not trip that check on the + * across-partitions property that comes with it. + */ + @SuppressWarnings("deprecation") + @Test + public void testLoadConfWithAPositiveMaxPendingMessages() { + producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); + producerBuilderImpl.loadConf(Map.of("maxPendingMessages", 5000)); + + assertEquals(producerBuilderImpl.getConf().getMaxPendingMessages(), 5000); + } + @Test public void testProducerBuilderImplWhenMessageRoutingIsSetImplicitly() throws PulsarClientException { producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); @@ -378,11 +394,25 @@ public void testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropert @SuppressWarnings("deprecation") @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = - "maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages") + "maxPendingMessagesAcrossPartitions needs to be >= 0") public void testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropertyIsInvalidErrorMessages() { producerBuilderImpl.maxPendingMessagesAcrossPartitions(-1); } + /** + * The across-partitions budget is allowed to be below {@code maxPendingMessages}: it is a budget + * shared by every partition, and the per-partition limit is lowered to its share where it is used. + * Rejecting it here made the two setters order-dependent. + */ + @SuppressWarnings("deprecation") + @Test + public void testAcrossPartitionsLimitBelowMaxPendingMessagesIsAccepted() { + producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); + producerBuilderImpl.maxPendingMessages(1000).maxPendingMessagesAcrossPartitions(500); + + assertEquals(producerBuilderImpl.getConf().getMaxPendingMessagesAcrossPartitions(), 500); + } + @SuppressWarnings("deprecation") @Test public void testProducerBuilderImplWhenNumericPropertiesAreValid() { From 9390c145eed384403f51eb7811f15ee299a2388b Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 18 Aug 2026 15:38:42 +0300 Subject: [PATCH 3/4] [fix][client] Apply the no-memory-limit producer queue defaults only where unset ### Motivation When the client memory limit is disabled there is no byte-based backpressure left, so #15723 gave producers the pre-PIP-120 pending-message defaults instead of letting them buffer without any limit. Its commit message states the intent: "restore maxPendingMessages and maxPendingMessagesAcrossPartitions when memory limit is disabled so that pre-PIP-120 default configuration is restored when limit is disabled". They are defaults, and PIP-120 is the same commit that changed them to 0 and documented 0 as "disable the pending messages check". An application that configures a limit therefore has to keep winning over them, including with an explicit 0. The previous approach seeded the defaults onto the builder in `newProducer(Schema)`, which left three holes: 1. The no-argument `newProducer()` never got them, so the WebSocket proxy's producer handler, the Functions log appender and the Functions worker are unbounded today. 2. A caller that later passed a limit through - a CLI flag or a config value sitting at its own default - silently replaced them. 3. Seeding `maxPendingMessages` to 1000 made a later `maxPendingMessagesAcrossPartitions(500)` throw, which a Functions `ProducerConfig` setting only that limit walks straight into. ### Modifications `maxPendingMessages` is a primitive whose unset value is 0, and 0 is also a meaningful explicit value, so the configuration alone cannot tell "never configured" from "explicitly unbounded". `ProducerBuilderImpl` records which of the two limits the application configured - through the setters or through `loadConf` - and carries that across `clone()`. At producer creation the client fills in only the limits that were never configured, on a copy of the configuration so nothing leaks into the next producer built from the same builder. A limit that is already positive counts as configured however the configuration was populated. Tracking this on the builder rather than on the configuration is not a preference: `loadConf` goes through `ConfigurationDataUtils.loadData`, which rebuilds the configuration by replaying every property through the setters, so a marker held there would be re-set on every call. An explicit `maxPendingMessages(0)` also suppresses the across-partitions default, so one call is enough to ask for a producer with no message-count limit whatever the topic's shape - filling in the budget would put a per-partition limit straight back. The V5 client is deliberately left out. It reaches producer creation through `createSegmentProducerAsync` and exposes no pending-message setting at all, so its client memory limit is the only backpressure it has, and the only thing an application can turn off. The WebSocket proxy takes `maxPendingMessages` from a query parameter and its client has no memory limit by default, so a remote client could now ask for an unbounded pending queue inside the shared proxy. A non-positive value is ignored there. Assisted-by: Claude Code (Opus 5) --- .../client/api/ProducerQueueSizeTest.java | 114 ++++++++++++++++-- .../pulsar/client/api/ProducerBuilder.java | 16 ++- .../client/impl/ProducerBuilderImpl.java | 27 ++++- .../pulsar/client/impl/PulsarClientImpl.java | 81 ++++++++----- .../client/impl/ProducerBuilderImplTest.java | 11 +- .../functions/instance/ContextImplTest.java | 5 + .../pulsar/websocket/ProducerHandler.java | 12 +- .../AbstractWebSocketHandlerTest.java | 19 ++- 8 files changed, 237 insertions(+), 48 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java index 11e054ac99f0a..cc4b53d101705 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import lombok.Cleanup; import org.apache.pulsar.broker.service.SharedPulsarBaseTest; @@ -43,6 +44,11 @@ private static ProducerConfigurationData confOf(Producer producer) { return ((ProducerBase) producer).getConfiguration(); } + @DataProvider(name = "partitioned") + public Object[][] partitioned() { + return new Object[][]{{Boolean.FALSE}, {Boolean.TRUE}}; + } + @DataProvider(name = "matrix") public Object[][] matrix() { return new Object[][]{ @@ -111,14 +117,43 @@ public void testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled() throws Except } /** - * Setting the pending-message limits to 0 after the builder has been created must not leave a - * producer with no bound at all when the client memory limit is disabled: 0 means "unset", and - * unset falls back to the bounded defaults. + * The fallback is a default, not a floor. An application that asks for no message-count limit at + * all still gets it, by passing 0 explicitly. This is what keeps 0 a usable value rather than an + * alias for "unset". + * + *

A single {@code maxPendingMessages(0)} has to be enough whatever the topic's shape: filling + * in the across-partitions budget would put a per-partition limit back on a partitioned topic. + */ + @Test(dataProvider = "partitioned") + public void testExplicitZeroDisablesTheBoundWhenMemoryLimitDisabled(boolean partitioned) throws Exception { + String topic = newTopicName(); + if (partitioned) { + admin.topics().createPartitionedTopic(topic, 10); + } + + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer(Schema.BYTES) + .topic(topic) + .maxPendingMessages(0) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } + + /** + * Mirror of the above: disabling only the across-partitions budget must not take the per-producer + * default down with it. Filling in that default would otherwise be capped by a budget of 0. */ @SuppressWarnings("deprecation") @Test - public void testLateZeroMaxPendingMessagesDoesNotDisableTheBoundWhenMemoryLimitDisabled() - throws Exception { + public void testExplicitZeroAcrossPartitionsKeepsThePerProducerDefault() throws Exception { @Cleanup PulsarClient client = PulsarClient.builder() .serviceUrl(getWebServiceUrl()) @@ -126,18 +161,83 @@ public void testLateZeroMaxPendingMessagesDoesNotDisableTheBoundWhenMemoryLimitD .build(); @Cleanup - Producer producer = client.newProducer(Schema.BYTES) + Producer producer = client.newProducer() .topic(newTopicName()) - .maxPendingMessages(0) .maxPendingMessagesAcrossPartitions(0) .create(); + assertThat(confOf(producer).getMaxPendingMessages()) + .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } + + /** + * {@code loadConf} is the other way an application configures a limit. A limit present in the map + * counts as configured, including a 0, even though {@code loadConf} rebuilds the configuration + * object and so cannot carry any marker on it. + */ + @Test + public void testLoadConfZeroDisablesTheBoundWhenMemoryLimitDisabled() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .loadConf(Map.of("maxPendingMessages", 0)) + .create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero(); + } + + /** + * A {@code loadConf} that does not mention the limits leaves them unconfigured, so the defaults + * still apply. Pins that rebuilding the configuration is not mistaken for configuring it. + */ + @Test + public void testLoadConfWithoutTheLimitsKeepsTheDefaults() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + @Cleanup + Producer producer = client.newProducer() + .topic(newTopicName()) + .loadConf(Map.of("producerName", "loadConfWithoutLimits")) + .create(); + assertThat(confOf(producer).getMaxPendingMessages()) .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES); assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()) .isEqualTo(NO_MEMORY_LIMIT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); } + /** + * A cloned builder has to keep knowing which limits were configured, or the clone would silently + * get the defaults back. + */ + @Test + public void testCloneKeepsAnExplicitlyDisabledBound() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder() + .serviceUrl(getWebServiceUrl()) + .memoryLimit(0, SizeUnit.BYTES) + .build(); + + ProducerBuilder builder = client.newProducer().maxPendingMessages(0); + + @Cleanup + Producer producer = builder.clone().topic(newTopicName()).create(); + + assertThat(confOf(producer).getMaxPendingMessages()).isZero(); + } + /** * {@code maxPendingMessagesAcrossPartitions} must be {@code >= maxPendingMessages}. Filling in * the across-partitions fallback must therefore never lower it below an explicitly configured diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java index 1cb6db79e9313..4e37bb4d19809 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java @@ -176,9 +176,13 @@ public interface ProducerBuilder extends Cloneable { * *

Default is 0, which disables the pending messages check. Disabling it only removes the * message-count limit; the memory the pending queue may hold is then bounded by the client - * memory limit ({@link ClientBuilder#memoryLimit(long, SizeUnit)}) instead. When the client - * memory limit is also disabled there would be no backpressure left at all, so producers fall - * back to a default pending messages queue size rather than buffering without limit. + * memory limit ({@link ClientBuilder#memoryLimit(long, SizeUnit)}) instead. + * + *

On a client whose memory limit is disabled there would be no backpressure left at all, so a + * producer that does not configure this setting falls back to a default queue size of 1000 rather + * than buffering without limit. Calling this method always wins over that default, so passing 0 + * explicitly is how an application asks for a producer with no message-count limit, on a + * partitioned topic as well. * * @param maxPendingMessages * the max size of the pending messages queue for the producer @@ -195,9 +199,9 @@ public interface ProducerBuilder extends Cloneable { * of pending messages when publishing on a partitioned topic. * *

Default is 0, which disables the pending messages across partitions check. As with - * {@link #maxPendingMessages(int)}, a producer created on a client whose memory limit is - * disabled falls back to a default budget instead, since no backpressure would otherwise be - * left. + * {@link #maxPendingMessages(int)}, a producer that does not configure this setting on a client + * whose memory limit is disabled falls back to a default budget of 50000 instead, since no + * backpressure would otherwise be left, and calling this method always wins over that default. * *

If publishing at a high rate over a topic with many partitions (especially when publishing messages without a * partitioning key), it might be beneficial to increase this parameter to allow for more pipelining within the diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java index 9242cfd6a08cf..11915140ceab9 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java @@ -56,6 +56,14 @@ public class ProducerBuilderImpl implements ProducerBuilder { private ProducerConfigurationData conf; private Schema schema; private List interceptorList; + /** + * Whether the application configured the pending-message limits. Their unset value is 0, which is + * also a meaningful explicit value ("no message-count limit"), so the configuration alone cannot + * tell the two apart. See + * {@link PulsarClientImpl#applyNoMemoryLimitProducerDefaults(ProducerConfigurationData, boolean, boolean)}. + */ + private boolean maxPendingMessagesConfigured; + private boolean maxPendingMessagesAcrossPartitionsConfigured; public ProducerBuilderImpl(PulsarClientImpl client, Schema schema) { this(client, new ProducerConfigurationData(), schema); @@ -78,7 +86,10 @@ public ProducerBuilder schema(Schema schema) { @Override public ProducerBuilder clone() { - return new ProducerBuilderImpl<>(client, conf.clone(), schema); + ProducerBuilderImpl copy = new ProducerBuilderImpl<>(client, conf.clone(), schema); + copy.maxPendingMessagesConfigured = maxPendingMessagesConfigured; + copy.maxPendingMessagesAcrossPartitionsConfigured = maxPendingMessagesAcrossPartitionsConfigured; + return copy; } @Override @@ -120,15 +131,23 @@ public CompletableFuture> createAsync() { client.instrumentProvider())); } + ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf, + maxPendingMessagesConfigured, maxPendingMessagesAcrossPartitionsConfigured); + return effectiveInterceptors == null || effectiveInterceptors.size() == 0 - ? client.createProducerAsync(conf, schema, null) - : client.createProducerAsync(conf, schema, new ProducerInterceptors(effectiveInterceptors)); + ? client.createProducerAsync(producerConf, schema, null) + : client.createProducerAsync(producerConf, schema, new ProducerInterceptors(effectiveInterceptors)); } @Override public ProducerBuilder loadConf(Map config) { conf = ConfigurationDataUtils.loadData( config, conf, ProducerConfigurationData.class); + // A limit present in the map was configured by the application, even when its value is the + // same as the unset one. loadData builds a new configuration instance, so this cannot be + // tracked in the configuration itself. + maxPendingMessagesConfigured |= config.containsKey("maxPendingMessages"); + maxPendingMessagesAcrossPartitionsConfigured |= config.containsKey("maxPendingMessagesAcrossPartitions"); return this; } @@ -154,6 +173,7 @@ public ProducerBuilder sendTimeout(int sendTimeout, @NonNull TimeUnit unit) { @Override public ProducerBuilder maxPendingMessages(int maxPendingMessages) { conf.setMaxPendingMessages(maxPendingMessages); + maxPendingMessagesConfigured = true; return this; } @@ -161,6 +181,7 @@ public ProducerBuilder maxPendingMessages(int maxPendingMessages) { @Override public ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) { conf.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); + maxPendingMessagesAcrossPartitionsConfigured = true; return this; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index f30ba14fa4cd2..221c75436c063 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -693,61 +693,84 @@ public CompletableFuture> createProducerAsync(ProducerConfigurat + " Topic: '" + topic + "'")); } - final ProducerConfigurationData producerConf = resolvePendingMessagesLimits(conf); - if (schema instanceof AutoProduceBytesSchema) { AutoProduceBytesSchema autoProduceBytesSchema = (AutoProduceBytesSchema) schema; if (autoProduceBytesSchema.hasUserProvidedSchema()) { - return createProducerAsync(topic, producerConf, schema, interceptors); + return createProducerAsync(topic, conf, schema, interceptors); } return reloadSchemaForAutoProduceProducer(topic, autoProduceBytesSchema) - .thenCompose(schemaInfoOptional -> - createProducerAsync(topic, producerConf, schema, interceptors)); + .thenCompose(schemaInfoOptional -> createProducerAsync(topic, conf, schema, interceptors)); } else { - return createProducerAsync(topic, producerConf, schema, interceptors); + return createProducerAsync(topic, conf, schema, interceptors); } } /** - * Resolve the producer's pending-message limits before the producer is created. + * Apply the default pending-message limits a producer gets when this client has no memory limit. * *

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. When it is disabled - * there is nothing left to bound that queue, so fall back to message-count limits rather than - * let a producer that outruns its broker buffer without any limit at all. + * there is nothing left to bound that queue, so producers fall back to the pre-PIP-120 + * message-count defaults rather than buffering without any limit at all. + * + *

These are defaults, not a floor. A limit the application configured is always kept — including + * an explicit {@code 0}, which is how an application asks for no message-count limit at all. Only a + * limit that was never configured is filled in, which is why the caller passes in what it saw + * rather than letting this method infer it: {@code 0} is both the unset value and a meaningful + * explicit one. + * + *

Note that on a partitioned topic a filled-in across-partitions budget is still divided between + * the partitions afterwards, which can lower an explicitly configured per-producer limit. * - *

Only limits that are unset are filled in; an explicit limit is never overwritten. Note that - * on a partitioned topic a filled-in across-partitions budget is still divided between the - * partitions afterwards, which can lower an explicitly configured per-producer limit. Doing this - * at creation time rather than on the builder means the fallback cannot be missed depending on - * which {@code newProducer} overload produced the builder, nor undone by a later call setting a - * limit back to its unset value. Segment producers created for the V5 client go through - * {@link #createSegmentProducerAsync} and do not pass through here. + *

Called by {@link ProducerBuilderImpl}, which is what knows whether a limit was configured. The + * V5 client builds its segment producers through {@link #createSegmentProducerAsync} instead, and + * deliberately gets no defaults here: it exposes no pending-message setting at all, so its client + * memory limit is the only backpressure it has and the only thing an application can turn off. * * @param conf the requested producer configuration - * @return the configuration to create the producer with; a resolved copy when a fallback - * applies, otherwise {@code conf} unchanged + * @param maxPendingMessagesConfigured whether the application configured {@code maxPendingMessages} + * @param maxPendingMessagesAcrossPartitionsConfigured whether the application configured + * {@code maxPendingMessagesAcrossPartitions} + * @return the configuration to create the producer with; a resolved copy when a default applies, + * otherwise {@code conf} unchanged */ - private ProducerConfigurationData resolvePendingMessagesLimits(ProducerConfigurationData conf) { - if (memoryLimitController.isMemoryLimited() - || (conf.getMaxPendingMessages() > 0 && conf.getMaxPendingMessagesAcrossPartitions() > 0)) { + public ProducerConfigurationData applyNoMemoryLimitProducerDefaults(ProducerConfigurationData conf, + boolean maxPendingMessagesConfigured, boolean maxPendingMessagesAcrossPartitionsConfigured) { + // A limit that is already positive was configured by definition, whichever way the + // configuration was populated. The flags only tell an explicit 0 apart from an unset one. + maxPendingMessagesConfigured |= conf.getMaxPendingMessages() > 0; + maxPendingMessagesAcrossPartitionsConfigured |= conf.getMaxPendingMessagesAcrossPartitions() > 0; + if ((maxPendingMessagesConfigured && maxPendingMessagesAcrossPartitionsConfigured) + || memoryLimitController.isMemoryLimited()) { return conf; } - int maxPendingMessages = conf.getMaxPendingMessages() > 0 + int maxPendingMessages = maxPendingMessagesConfigured ? conf.getMaxPendingMessages() : NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES; - int maxPendingMessagesAcrossPartitions = conf.getMaxPendingMessagesAcrossPartitions() > 0 - ? conf.getMaxPendingMessagesAcrossPartitions() - : Math.max(maxPendingMessages, NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); - // The across-partitions limit is a budget shared by every partition, so a single producer's - // queue can never exceed it. This also keeps the two limits consistent for the setters below. - maxPendingMessages = Math.min(maxPendingMessages, maxPendingMessagesAcrossPartitions); + final int maxPendingMessagesAcrossPartitions; + if (maxPendingMessagesAcrossPartitionsConfigured) { + maxPendingMessagesAcrossPartitions = conf.getMaxPendingMessagesAcrossPartitions(); + } else if (maxPendingMessages == 0) { + // The application configured no per-producer limit. Filling in a partitions budget would + // put one back, because a partitioned producer derives its per-partition limit from it, so + // a single maxPendingMessages(0) is enough to ask for a producer with no message-count + // limit whatever the topic's shape. + maxPendingMessagesAcrossPartitions = 0; + } else { + maxPendingMessagesAcrossPartitions = + Math.max(maxPendingMessages, NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); + } + if (maxPendingMessagesAcrossPartitions > 0) { + // The across-partitions limit is a budget shared by every partition, so a single producer's + // queue can never exceed it. A configured 0 means there is no such budget and is left + // alone, rather than capping every producer at zero. + maxPendingMessages = Math.min(maxPendingMessages, maxPendingMessagesAcrossPartitions); + } // Resolve on a copy: the builder hands over its own configuration instance, so filling in a // limit here would otherwise leak into the next producer built from the same builder. ProducerConfigurationData resolved = conf.clone(); - // Order matters: the across-partitions setter rejects a value below maxPendingMessages. resolved.setMaxPendingMessages(maxPendingMessages); resolved.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); log.debug().attr("topic", conf.getTopicName()) diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java index 86468b4badb97..bdae20c1ed22e 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java @@ -19,11 +19,13 @@ package org.apache.pulsar.client.impl; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyBoolean; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @@ -64,6 +66,11 @@ public void setup() { producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES); when(client.newProducer()).thenReturn(producerBuilderImpl); + // The builder asks the client to fill in the pending-message defaults before creating the + // producer; on a mock that would otherwise hand back a null configuration. + when(client.applyNoMemoryLimitProducerDefaults(any(ProducerConfigurationData.class), anyBoolean(), + anyBoolean())).thenAnswer(invocation -> invocation.getArgument(0)); + doReturn(CompletableFuture.completedFuture(producer)) .when(client).createProducerAsync( any(ProducerConfigurationData.class), any(), eq(null)); @@ -124,7 +131,7 @@ public void testProducerBuilderImplWhenMessageRoutingModeIsRoundRobinPartition() * {@code loadConf} rebuilds the configuration by replaying every property through the public * setters, and {@code setMaxPendingMessagesAcrossPartitions} rejects a value below * {@code maxPendingMessages}. Pins that loading a positive limit does not trip that check on the - * across-partitions property that comes with it. + * across-partitions property that comes with it, and that the limit is recorded as configured. */ @SuppressWarnings("deprecation") @Test @@ -133,6 +140,8 @@ public void testLoadConfWithAPositiveMaxPendingMessages() { producerBuilderImpl.loadConf(Map.of("maxPendingMessages", 5000)); assertEquals(producerBuilderImpl.getConf().getMaxPendingMessages(), 5000); + assertTrue(producerBuilderImpl.isMaxPendingMessagesConfigured()); + assertFalse(producerBuilderImpl.isMaxPendingMessagesAcrossPartitionsConfigured()); } @Test diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java index f20fc07a5907d..d7fd8ad19979a 100644 --- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java +++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java @@ -20,6 +20,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.same; @@ -111,6 +112,10 @@ public void setup() throws PulsarClientException { when(client.newProducer()).thenAnswer(invocation -> new ProducerBuilderImpl<>(client, Schema.BYTES)); when(client.newProducer(any())).thenAnswer( invocation -> new ProducerBuilderImpl<>(client, invocation.getArgument(0))); + // The builder asks the client to fill in the pending-message defaults before creating the + // producer; on a mock that would otherwise hand back a null configuration. + when(client.applyNoMemoryLimitProducerDefaults(any(ProducerConfigurationData.class), anyBoolean(), + anyBoolean())).thenAnswer(invocation -> invocation.getArgument(0)); when(client.createProducerAsync(any(ProducerConfigurationData.class), any(), any())) .thenReturn(CompletableFuture.completedFuture(producer)); when(client.getSchema(anyString())).thenReturn(CompletableFuture.completedFuture(Optional.empty())); diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java index 2d1f46cc47427..89ad8dc48b305 100644 --- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java +++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java @@ -508,7 +508,17 @@ private void popularProducerBuilderForServerSideEncrypt(ProducerBuilder } if (queryParams.containsKey("maxPendingMessages")) { - builder.maxPendingMessages(Integer.parseInt(queryParams.get("maxPendingMessages"))); + int maxPendingMessages = Integer.parseInt(queryParams.get("maxPendingMessages")); + if (maxPendingMessages > 0) { + builder.maxPendingMessages(maxPendingMessages); + } else { + // 0 asks the client for an unbounded pending queue. The proxy's client runs + // without a memory limit by default (webSocketPulsarClientMemoryLimitInMB), + // so that would leave this producer with no backpressure at all, buffering a + // remote client's messages in the shared proxy. Keep the client's default. + log.info().attr("maxPendingMessages", maxPendingMessages) + .log("Ignoring the param maxPendingMessages of producer since it is not positive"); + } } if (queryParams.containsKey("batchingMaxPublishDelay")) { diff --git a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java index aa0ed89c72cb7..51648a974d9db 100644 --- a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java +++ b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java @@ -214,8 +214,13 @@ public MockedProducerHandler(WebSocketService service, HttpServletRequest reques super(service, request, response); } + @SuppressWarnings("unchecked") + public ProducerBuilderImpl getBuilder() throws PulsarClientException { + return (ProducerBuilderImpl) getProducerBuilder(newPulsarClient()); + } + public ProducerConfigurationData getConf() throws PulsarClientException { - return ((ProducerBuilderImpl) getProducerBuilder(newPulsarClient())).getConf(); + return getBuilder().getConf(); } public void clearQueryParams() { @@ -283,6 +288,18 @@ public void producerBuilderTest() throws IOException { conf = producerHandler.getConf(); // ProducerHandler doesn't support CustomPartition assertEquals(conf.getMessageRoutingMode(), MessageRoutingMode.SinglePartition); + + // A maxPendingMessages of 0 asks the client for an unbounded pending queue. The proxy's client + // runs without a memory limit by default, so honouring it would leave a remote client's + // producer with no backpressure at all inside the shared proxy. It is ignored instead, which + // leaves the limit unconfigured so that the client's own default applies. + producerHandler.clearQueryParams(); + producerHandler.putQueryParam("batchingEnabled", "true"); + producerHandler.putQueryParam("maxPendingMessages", "0"); + assertFalse(producerHandler.getBuilder().isMaxPendingMessagesConfigured()); + + producerHandler.putQueryParam("maxPendingMessages", "1001"); + assertTrue(producerHandler.getBuilder().isMaxPendingMessagesConfigured()); } class MockedConsumerHandler extends ConsumerHandler { From 16815fd74fec0c757255e90c7aa571611c6e1dbb Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 18 Aug 2026 21:45:17 +0300 Subject: [PATCH 4/4] [fix][client] Revert the WebSocket proxy guard, handled separately The guard on the proxy's remote-supplied maxPendingMessages query parameter belongs with giving the proxy a memory limit, which is a change of its own: `webSocketPulsarClientMemoryLimitInMB` is an `int` defaulting to 0, so the proxy always overrides the client's own 64M default with "no limit". Bounding the proxy by bytes is the better fit there - it multiplexes many producers over one client - and it makes the query parameter a separate question rather than the only defence. Assisted-by: Claude Code (Opus 5) --- .../pulsar/websocket/ProducerHandler.java | 12 +----------- .../AbstractWebSocketHandlerTest.java | 19 +------------------ 2 files changed, 2 insertions(+), 29 deletions(-) diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java index 89ad8dc48b305..2d1f46cc47427 100644 --- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java +++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java @@ -508,17 +508,7 @@ private void popularProducerBuilderForServerSideEncrypt(ProducerBuilder } if (queryParams.containsKey("maxPendingMessages")) { - int maxPendingMessages = Integer.parseInt(queryParams.get("maxPendingMessages")); - if (maxPendingMessages > 0) { - builder.maxPendingMessages(maxPendingMessages); - } else { - // 0 asks the client for an unbounded pending queue. The proxy's client runs - // without a memory limit by default (webSocketPulsarClientMemoryLimitInMB), - // so that would leave this producer with no backpressure at all, buffering a - // remote client's messages in the shared proxy. Keep the client's default. - log.info().attr("maxPendingMessages", maxPendingMessages) - .log("Ignoring the param maxPendingMessages of producer since it is not positive"); - } + builder.maxPendingMessages(Integer.parseInt(queryParams.get("maxPendingMessages"))); } if (queryParams.containsKey("batchingMaxPublishDelay")) { diff --git a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java index 51648a974d9db..aa0ed89c72cb7 100644 --- a/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java +++ b/pulsar-websocket/src/test/java/org/apache/pulsar/websocket/AbstractWebSocketHandlerTest.java @@ -214,13 +214,8 @@ public MockedProducerHandler(WebSocketService service, HttpServletRequest reques super(service, request, response); } - @SuppressWarnings("unchecked") - public ProducerBuilderImpl getBuilder() throws PulsarClientException { - return (ProducerBuilderImpl) getProducerBuilder(newPulsarClient()); - } - public ProducerConfigurationData getConf() throws PulsarClientException { - return getBuilder().getConf(); + return ((ProducerBuilderImpl) getProducerBuilder(newPulsarClient())).getConf(); } public void clearQueryParams() { @@ -288,18 +283,6 @@ public void producerBuilderTest() throws IOException { conf = producerHandler.getConf(); // ProducerHandler doesn't support CustomPartition assertEquals(conf.getMessageRoutingMode(), MessageRoutingMode.SinglePartition); - - // A maxPendingMessages of 0 asks the client for an unbounded pending queue. The proxy's client - // runs without a memory limit by default, so honouring it would leave a remote client's - // producer with no backpressure at all inside the shared proxy. It is ignored instead, which - // leaves the limit unconfigured so that the client's own default applies. - producerHandler.clearQueryParams(); - producerHandler.putQueryParam("batchingEnabled", "true"); - producerHandler.putQueryParam("maxPendingMessages", "0"); - assertFalse(producerHandler.getBuilder().isMaxPendingMessagesConfigured()); - - producerHandler.putQueryParam("maxPendingMessages", "1001"); - assertTrue(producerHandler.getBuilder().isMaxPendingMessagesConfigured()); } class MockedConsumerHandler extends ConsumerHandler {