[fix][client] Apply no-memory-limit producer queue defaults at producer creation

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

@@ -18,16 +18,37 @@
  */
 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.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 = "partitioned")
+    public Object[][] partitioned() {
+        return new Object[][]{{Boolean.FALSE}, {Boolean.TRUE}};
+    }
+
     @DataProvider(name = "matrix")
     public Object[][] matrix() {
         return new Object[][]{
@@ -72,4 +93,277 @@ 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<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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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(getWebServiceUrl())
+                .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();
+    }
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java

@@ -174,7 +174,15 @@
      * 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 @@
      * 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
@@ -561,7 +572,7 @@
      * @return the producer builder instance
      */
     @Deprecated
     ProducerBuilder<T> intercept(ProducerInterceptor<T> ... interceptors);
 
     /**
      * Add a set of {@link org.apache.pulsar.client.api.interceptor.ProducerInterceptor} to the producer.