You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/08/19 09:25:25 UTC
(pulsar) 01/02: [fix][client] Apply no-memory-limit producer queue defaults at producer creation (#26342)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4c9e045854eda57e6963ee082752feda04712ce6
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Aug 19 01:28:52 2026 +0300
[fix][client] Apply no-memory-limit producer queue defaults at producer creation (#26342)
(cherry picked from commit b660d70566d0e049b3a459a272874b8a47a41c09)
---
.../pulsar/client/api/ProducerQueueSizeTest.java | 294 +++++++++++++++++++++
.../apache/pulsar/client/api/ProducerBuilder.java | 15 +-
.../pulsar/client/impl/ProducerBuilderImpl.java | 27 +-
.../pulsar/client/impl/PulsarClientImpl.java | 81 +++++-
.../impl/conf/ProducerConfigurationData.java | 12 +-
.../client/impl/ProducerBuilderImplTest.java | 42 ++-
.../pulsar/functions/instance/ContextImplTest.java | 5 +
7 files changed, 460 insertions(+), 16 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 0c470ae58751..33badf3e52fe 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,10 +18,14 @@
*/
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.Map;
import java.util.concurrent.CompletableFuture;
import lombok.Cleanup;
+import org.apache.pulsar.client.impl.ProducerBase;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -29,6 +33,14 @@ import org.testng.annotations.Test;
public class ProducerQueueSizeTest extends ProducerConsumerBase {
+ /**
+ * 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;
+
@BeforeMethod
@Override
protected void setup() throws Exception {
@@ -42,6 +54,15 @@ public class ProducerQueueSizeTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ 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[][]{
@@ -85,4 +106,277 @@ public class ProducerQueueSizeTest extends ProducerConsumerBase {
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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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);
+ }
+
+ /**
+ * 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".
+ *
+ * <p>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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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 testExplicitZeroAcrossPartitionsKeepsThePerProducerDefault() throws Exception {
+ @Cleanup
+ PulsarClient client = PulsarClient.builder()
+ .serviceUrl(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic(newTopicName())
+ .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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ ProducerBuilder<byte[]> builder = client.newProducer().maxPendingMessages(0);
+
+ @Cleanup
+ Producer<byte[]> 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
+ * 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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ ProducerBuilder<byte[]> builder = client.newProducer();
+
+ @Cleanup
+ Producer<byte[]> 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<byte[]> 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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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(brokerUrl.toString())
+ .memoryLimit(0, SizeUnit.BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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(brokerUrl.toString())
+ .memoryLimit(64, SizeUnit.MEGA_BYTES)
+ .build();
+
+ @Cleanup
+ Producer<byte[]> 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 6015bc7af217..ec9d93dbfea7 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,15 @@ public interface ProducerBuilder<T> 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.
*
- * <p>Default is 0, which disables the pending messages check.
+ * <p>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.
+ *
+ * <p>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
@@ -190,7 +198,10 @@ public interface ProducerBuilder<T> 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.
*
- * <p>Default is 0, which disables the pending messages across partitions check.
+ * <p>Default is 0, which disables the pending messages across partitions check. As with
+ * {@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.
*
* <p>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 3457aac0402f..6d6c2081e2dd 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<T> implements ProducerBuilder<T> {
private ProducerConfigurationData conf;
private Schema<T> schema;
private List<ProducerInterceptor> 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<T> schema) {
this(client, new ProducerConfigurationData(), schema);
@@ -78,7 +86,10 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> clone() {
- return new ProducerBuilderImpl<>(client, conf.clone(), schema);
+ ProducerBuilderImpl<T> copy = new ProducerBuilderImpl<>(client, conf.clone(), schema);
+ copy.maxPendingMessagesConfigured = maxPendingMessagesConfigured;
+ copy.maxPendingMessagesAcrossPartitionsConfigured = maxPendingMessagesAcrossPartitionsConfigured;
+ return copy;
}
@Override
@@ -108,15 +119,23 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
return FutureUtil.failedFuture(pce);
}
+ ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf,
+ maxPendingMessagesConfigured, maxPendingMessagesAcrossPartitionsConfigured);
+
return interceptorList == null || interceptorList.size() == 0
- ? client.createProducerAsync(conf, schema, null)
- : client.createProducerAsync(conf, schema, new ProducerInterceptors(interceptorList));
+ ? client.createProducerAsync(producerConf, schema, null)
+ : client.createProducerAsync(producerConf, schema, new ProducerInterceptors(interceptorList));
}
@Override
public ProducerBuilder<T> loadConf(Map<String, Object> 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;
}
@@ -142,6 +161,7 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> maxPendingMessages(int maxPendingMessages) {
conf.setMaxPendingMessages(maxPendingMessages);
+ maxPendingMessagesConfigured = true;
return this;
}
@@ -149,6 +169,7 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public ProducerBuilder<T> 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 5f9acf2ff313..fd626c24d471 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
@@ -305,14 +305,7 @@ public class PulsarClientImpl implements PulsarClient {
@Override
public <T> ProducerBuilder<T> newProducer(Schema<T> schema) {
- ProducerBuilderImpl<T> 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
@@ -401,6 +394,78 @@ public class PulsarClientImpl implements PulsarClient {
}
+ /**
+ * Apply the default pending-message limits a producer gets when this client has no memory limit.
+ *
+ * <p>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 producers fall back to the pre-PIP-120
+ * message-count defaults rather than buffering without any limit at all.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>Called by {@link ProducerBuilderImpl}, which is what knows whether a limit was configured.
+ *
+ * @param conf the requested producer configuration
+ * @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
+ */
+ 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 = maxPendingMessagesConfigured
+ ? conf.getMaxPendingMessages()
+ : NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES;
+ 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();
+ resolved.setMaxPendingMessages(maxPendingMessages);
+ resolved.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Client memory limit is disabled, applying default producer pending message limits."
+ + " maxPendingMessages: {}, maxPendingMessagesAcrossPartitions: {}",
+ conf.getTopicName(), maxPendingMessages, maxPendingMessagesAcrossPartitions);
+ }
+ return resolved;
+ }
+
public CompletableFuture<Void> reloadSchemaForAutoProduceProducer(String topic, AutoProduceBytesSchema autoSchema) {
return lookup.getSchema(TopicName.get(topic)).thenAccept(schemaInfoOptional -> {
if (schemaInfoOptional.isPresent()) {
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 a7da82599398..cec15904f71a 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
@@ -250,9 +250,17 @@ public class ProducerConfigurationData implements Serializable, Cloneable {
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 a49114d67b58..6033ebc1146e 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,12 +19,16 @@
package org.apache.pulsar.client.impl;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyBoolean;
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;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -60,6 +64,11 @@ public class ProducerBuilderImplTest {
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));
+
when(client.createProducerAsync(
any(ProducerConfigurationData.class), any(Schema.class), eq(null)))
.thenReturn(CompletableFuture.completedFuture(producer));
@@ -116,6 +125,23 @@ public class ProducerBuilderImplTest {
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, and that the limit is recorded as configured.
+ */
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testLoadConfWithAPositiveMaxPendingMessages() {
+ producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES);
+ producerBuilderImpl.loadConf(Collections.singletonMap("maxPendingMessages", 5000));
+
+ assertEquals(producerBuilderImpl.getConf().getMaxPendingMessages(), 5000);
+ assertTrue(producerBuilderImpl.isMaxPendingMessagesConfigured());
+ assertFalse(producerBuilderImpl.isMaxPendingMessagesAcrossPartitionsConfigured());
+ }
+
@Test
public void testProducerBuilderImplWhenMessageRoutingIsSetImplicitly() throws PulsarClientException {
producerBuilderImpl = new ProducerBuilderImpl(client, Schema.BYTES);
@@ -373,11 +399,25 @@ public class ProducerBuilderImplTest {
}
@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);
+ }
+
@Test
public void testProducerBuilderImplWhenNumericPropertiesAreValid() {
producerBuilderImpl.batchingMaxPublishDelay(1, TimeUnit.SECONDS);
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 cb4c93f153fd..071886698fb4 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 @@ package org.apache.pulsar.functions.instance;
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;
@@ -108,6 +109,10 @@ public class ContextImplTest {
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()));