[fix][client] Divide the across-partitions budget only when it was set explicitly

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerQueueSizeTest.java

@@ -366,4 +366,107 @@ public void testMemoryLimitedClientKeepsUnboundedPendingMessages() throws Except
         assertThat(confOf(producer).getMaxPendingMessages()).isZero();
         assertThat(confOf(producer).getMaxPendingMessagesAcrossPartitions()).isZero();
     }
+
+    /**
+     * A budget the application asked for is what gets divided between the partitions. Pins that the
+     * per-producer default filled in alongside it does not win the division and cap every partition at
+     * that default instead of at its share of the budget.
+     */
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testExplicitAcrossPartitionsBudgetIsDividedBetweenThePartitions() throws Exception {
+        String topic = newTopicName();
+        admin.topics().createPartitionedTopic(topic, 10);
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(getWebServiceUrl())
+                .memoryLimit(0, SizeUnit.BYTES)
+                .build();
+
+        @Cleanup
+        Producer<byte[]> producer = client.newProducer()
+                .topic(topic)
+                .maxPendingMessagesAcrossPartitions(60_000)
+                .create();
+
+        assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(6_000);
+    }
+
+    /**
+     * The mirror image: a budget that was only filled in as a default must not be divided, because
+     * dividing it would lower a per-producer limit the application did ask for.
+     */
+    @Test
+    public void testFilledInBudgetDoesNotLowerAnExplicitPerProducerLimit() throws Exception {
+        String topic = newTopicName();
+        admin.topics().createPartitionedTopic(topic, 10);
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(getWebServiceUrl())
+                .memoryLimit(0, SizeUnit.BYTES)
+                .build();
+
+        @Cleanup
+        Producer<byte[]> producer = client.newProducer()
+                .topic(topic)
+                .maxPendingMessages(60_000)
+                .create();
+
+        assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(60_000);
+    }
+
+    /**
+     * A budget smaller than the partition count divides to zero, which used to remove the queue bound
+     * altogether — asking for a tighter budget made the producer unbounded. Each partition keeps the
+     * smallest possible queue instead.
+     */
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testBudgetSmallerThanThePartitionCountStillBoundsEachPartition() throws Exception {
+        String topic = newTopicName();
+        admin.topics().createPartitionedTopic(topic, 10);
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(getWebServiceUrl())
+                .memoryLimit(0, SizeUnit.BYTES)
+                .build();
+
+        @Cleanup
+        Producer<byte[]> producer = client.newProducer()
+                .topic(topic)
+                .maxPendingMessagesAcrossPartitions(5)
+                .create();
+
+        assertThat(confOf(producer).getMaxPendingMessages()).isEqualTo(1);
+    }
+
+    /**
+     * An explicit {@code 0} means "no message-count limit" and stays that way on a partitioned topic,
+     * even next to an across-partitions budget. Reading the unset value instead of the marker used to
+     * overwrite it with the budget's per-partition share.
+     */
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testExplicitZeroIsKeptAlongsideAnAcrossPartitionsBudget() throws Exception {
+        String topic = newTopicName();
+        admin.topics().createPartitionedTopic(topic, 10);
+
+        @Cleanup
+        PulsarClient client = PulsarClient.builder()
+                .serviceUrl(getWebServiceUrl())
+                .memoryLimit(0, SizeUnit.BYTES)
+                .build();
+
+        @Cleanup
+        Producer<byte[]> producer = client.newProducer()
+                .topic(topic)
+                .maxPendingMessages(0)
+                .maxPendingMessagesAcrossPartitions(60_000)
+                .create();
+
+        assertThat(confOf(producer).getMaxPendingMessages()).isZero();
+    }
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java

@@ -19,8 +19,6 @@
 package org.apache.pulsar.client.impl;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES;
-import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
 import com.google.common.annotations.VisibleForTesting;
 import io.github.merlimat.slog.Logger;
 import io.netty.util.Timeout;
@@ -88,14 +86,23 @@ public PartitionedProducerImpl(PulsarClientImpl client, String topic, ProducerCo
                 ? new PartitionedTopicProducerStatsRecorderImpl()
                 : null;
 
+        // The across-partitions budget is a total shared by every partition, so it is divided between
+        // them here, where the partition count is finally known. Both limits are read through their
+        // "configured" markers rather than by comparing against their unset value, because that value
+        // is 0 for both and 0 is also a meaningful explicit setting.
         // MaxPendingMessagesAcrossPartitions doesn't support partial partition such as SinglePartition correctly
         int maxPendingMessages = conf.getMaxPendingMessages();
         int maxPendingMessagesAcrossPartitions = conf.getMaxPendingMessagesAcrossPartitions();
-        if (maxPendingMessagesAcrossPartitions != DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS) {
-            int maxPendingMsgsForOnePartition = maxPendingMessagesAcrossPartitions / numPartitions;
-            maxPendingMessages = (maxPendingMessages == DEFAULT_MAX_PENDING_MESSAGES)
-                    ? maxPendingMsgsForOnePartition
-                    : Math.min(maxPendingMessages, maxPendingMsgsForOnePartition);
+        if (conf.isMaxPendingMessagesAcrossPartitionsConfigured() && maxPendingMessagesAcrossPartitions > 0) {
+            // Never divide down to 0: a budget smaller than the partition count still asks for the
+            // smallest possible queue, not for the queue bound to be removed altogether.
+            int maxPendingMsgsForOnePartition =
+                    Math.max(1, maxPendingMessagesAcrossPartitions / numPartitions);
+            // A per-producer limit the application set wins, including an explicit 0 ("no message-count
+            // limit"). Only a limit it left unset — or one filled in as a default — adopts the share.
+            maxPendingMessages = conf.isMaxPendingMessagesConfigured()
+                    ? Math.min(maxPendingMessages, maxPendingMsgsForOnePartition)
+                    : maxPendingMsgsForOnePartition;
             conf.setMaxPendingMessages(maxPendingMessages);
         }
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java

@@ -56,14 +56,6 @@ 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);
@@ -86,10 +78,7 @@ public ProducerBuilder<T> schema(Schema<T> schema) {
 
     @Override
     public ProducerBuilder<T> clone() {
-        ProducerBuilderImpl<T> copy = new ProducerBuilderImpl<>(client, conf.clone(), schema);
-        copy.maxPendingMessagesConfigured = maxPendingMessagesConfigured;
-        copy.maxPendingMessagesAcrossPartitionsConfigured = maxPendingMessagesAcrossPartitionsConfigured;
-        return copy;
+        return new ProducerBuilderImpl<>(client, conf.clone(), schema);
     }
 
     @Override
@@ -131,8 +120,7 @@ public CompletableFuture<Producer<T>> createAsync() {
                     client.instrumentProvider()));
         }
 
-        ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf,
-                maxPendingMessagesConfigured, maxPendingMessagesAcrossPartitionsConfigured);
+        ProducerConfigurationData producerConf = client.applyNoMemoryLimitProducerDefaults(conf);
 
         return effectiveInterceptors == null || effectiveInterceptors.size() == 0
                 ? client.createProducerAsync(producerConf, schema, null)
@@ -141,13 +129,19 @@ public CompletableFuture<Producer<T>> createAsync() {
 
     @Override
     public ProducerBuilder<T> loadConf(Map<String, Object> config) {
+        // 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 by replaying every
+        // property through its setters, so the markers have to be carried over rather than read off
+        // the result.
+        boolean maxPendingMessagesConfigured =
+                conf.isMaxPendingMessagesConfigured() || config.containsKey("maxPendingMessages");
+        boolean maxPendingMessagesAcrossPartitionsConfigured =
+                conf.isMaxPendingMessagesAcrossPartitionsConfigured()
+                        || config.containsKey("maxPendingMessagesAcrossPartitions");
         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");
+        conf.setMaxPendingMessagesConfigured(maxPendingMessagesConfigured);
+        conf.setMaxPendingMessagesAcrossPartitionsConfigured(maxPendingMessagesAcrossPartitionsConfigured);
         return this;
     }
 
@@ -173,15 +167,13 @@ public ProducerBuilder<T> sendTimeout(int sendTimeout, @NonNull TimeUnit unit) {
     @Override
     public ProducerBuilder<T> maxPendingMessages(int maxPendingMessages) {
         conf.setMaxPendingMessages(maxPendingMessages);
-        maxPendingMessagesConfigured = true;
         return this;
     }
 
     @Deprecated
     @Override
     public ProducerBuilder<T> maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) {
         conf.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions);
-        maxPendingMessagesAcrossPartitionsConfigured = true;
         return this;
     }
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -199,7 +199,7 @@
 
     private final LoadingCache<String, SchemaInfoProvider> schemaProviderLoadingCache =
             CacheBuilder.newBuilder().maximumSize(100000)
                     .expireAfterAccess(30, TimeUnit.MINUTES)
                     .build(new CacheLoader<String, SchemaInfoProvider>() {
 
                         @Override
@@ -716,31 +716,32 @@
      *
      * <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.
+     * limit that was never configured is filled in, which is why this reads the markers the
+     * configuration carries rather than inferring it from the values: {@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>What is filled in here stays marked as unconfigured, so on a partitioned topic
+     * {@link PartitionedProducerImpl} divides only a budget the application actually asked for. A
+     * filled-in budget never lowers a per-producer limit that was asked for.
      *
-     * <p>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.
+     * <p>Called by {@link ProducerBuilderImpl}. 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
-     * @param maxPendingMessagesConfigured whether the application configured {@code maxPendingMessages}
-     * @param maxPendingMessagesAcrossPartitionsConfigured whether the application configured
-     *                                                     {@code maxPendingMessagesAcrossPartitions}
+     * @param conf the requested producer configuration, carrying the markers that say which limits the
+     *             application configured
      * @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) {
+    public ProducerConfigurationData applyNoMemoryLimitProducerDefaults(ProducerConfigurationData conf) {
         // 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;
+        // configuration was populated. The markers only tell an explicit 0 apart from an unset one.
+        boolean maxPendingMessagesConfigured =
+                conf.isMaxPendingMessagesConfigured() || conf.getMaxPendingMessages() > 0;
+        boolean maxPendingMessagesAcrossPartitionsConfigured =
+                conf.isMaxPendingMessagesAcrossPartitionsConfigured()
+                        || conf.getMaxPendingMessagesAcrossPartitions() > 0;
         if ((maxPendingMessagesConfigured && maxPendingMessagesAcrossPartitionsConfigured)
                 || memoryLimitController.isMemoryLimited()) {
             return conf;
@@ -752,10 +753,8 @@
         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.
+            // The application asked for no per-producer limit at all, so there is no queue for a
+            // budget to bound. Leaving it unset keeps the resolved configuration honest about that.
             maxPendingMessagesAcrossPartitions = 0;
         } else {
             maxPendingMessagesAcrossPartitions =
@@ -773,6 +772,12 @@
         ProducerConfigurationData resolved = conf.clone();
         resolved.setMaxPendingMessages(maxPendingMessages);
         resolved.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions);
+        // The setters mark whatever they are given as configured, so restore the markers: what is
+        // filled in here is a default, and a partitioned producer still has to be able to tell it apart
+        // from a limit the application asked for, so that a budget it did ask for is the one that gets
+        // divided.
+        resolved.setMaxPendingMessagesConfigured(maxPendingMessagesConfigured);
+        resolved.setMaxPendingMessagesAcrossPartitionsConfigured(maxPendingMessagesAcrossPartitionsConfigured);
         log.debug().attr("topic", conf.getTopicName())
                 .attr("maxPendingMessages", maxPendingMessages)
                 .attr("maxPendingMessagesAcrossPartitions", maxPendingMessagesAcrossPartitions)

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java

@@ -109,6 +109,30 @@ public class ProducerConfigurationData implements Serializable, Cloneable {
     )
     private int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
 
+    /**
+     * Whether the application configured {@link #maxPendingMessages}, and whether it configured
+     * {@link #maxPendingMessagesAcrossPartitions}.
+     *
+     * <p>The unset value of both limits is {@code 0}, which is also a meaningful explicit value ("no
+     * message-count limit" and "no across-partitions budget"), so the value alone cannot tell the two
+     * apart. Recording it here rather than on the builder lets everything that resolves these limits
+     * read the same answer, including {@code PartitionedProducerImpl}, which only sees the
+     * configuration.
+     *
+     * <p>Setting either limit through its setter marks it as configured, so a configuration populated
+     * directly rather than through {@code ProducerBuilderImpl} behaves the same way.
+     *
+     * <p>Deliberately not part of the serialized configuration: {@code loadConf} rebuilds the instance
+     * by replaying every property through its setters, which would mark both limits as configured
+     * whatever the application passed, so {@code ProducerBuilderImpl} restores them across that
+     * round-trip. {@code PulsarClientImpl} likewise restores them after filling in a default, which is
+     * not application input.
+     */
+    @JsonIgnore
+    private boolean maxPendingMessagesConfigured;
+    @JsonIgnore
+    private boolean maxPendingMessagesAcrossPartitionsConfigured;
+
     @Schema(
             name = "messageRoutingMode",
             description = "Message routing logic for producers on [partitioned topics]"
@@ -250,6 +274,7 @@ public void setProducerName(String producerName) {
     public void setMaxPendingMessages(int maxPendingMessages) {
         checkArgument(maxPendingMessages >= 0, "maxPendingMessages needs to be >= 0");
         this.maxPendingMessages = maxPendingMessages;
+        this.maxPendingMessagesConfigured = true;
     }
 
     /**
@@ -264,6 +289,7 @@ public void setMaxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPa
         checkArgument(maxPendingMessagesAcrossPartitions >= 0,
                 "maxPendingMessagesAcrossPartitions needs to be >= 0");
         this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions;
+        this.maxPendingMessagesAcrossPartitionsConfigured = true;
     }
 
     public void setBatchingMaxMessages(int batchingMaxMessages) {