[fix][cli][branch-4.2] Don't pass unset pulsar-perf pending-message options to the producer

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: whether a value is accepted depended on which of them
+     * had been called first, so setting only this one on a builder that already carries a default for
+     * the other throws. 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, and the budget means nothing on a non-partitioned topic.
+     */
     public void setMaxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) {
-        checkArgument(maxPendingMessagesAcrossPartitions >= maxPendingMessages,
-                "maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages");
+        checkArgument(maxPendingMessagesAcrossPartitions >= 0,
+                "maxPendingMessagesAcrossPartitions needs to be >= 0");
         this.maxPendingMessagesAcrossPartitions = maxPendingMessagesAcrossPartitions;
     }
 

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java

@@ -22,6 +22,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;
@@ -375,11 +376,25 @@ public void testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropert
     }
 
     @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, so setting only this one on a builder
+     * that already carried a default for the other threw.
+     */
+    @Test
+    public void testAcrossPartitionsLimitBelowMaxPendingMessagesIsAccepted() {
+        ProducerBuilderImpl<byte[]> builder = new ProducerBuilderImpl<>(client, Schema.BYTES);
+        builder.maxPendingMessages(1000).maxPendingMessagesAcrossPartitions(500);
+
+        assertEquals(builder.getConf().getMaxPendingMessagesAcrossPartitions(), 500);
+    }
+
     @Test
     public void testProducerBuilderImplWhenNumericPropertiesAreValid() {
         producerBuilderImpl.batchingMaxPublishDelay(1, TimeUnit.SECONDS);

pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java

@@ -22,8 +22,6 @@
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_BATCHING_MAX_MESSAGES;
-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.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
 import com.google.common.collect.Range;
@@ -64,6 +62,7 @@
 import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.client.api.transaction.Transaction;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
@@ -134,12 +133,13 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
             + "larger than allowed max size")
     private boolean chunkingAllowed = false;
 
-    @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding messages")
-    public int maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES;
+    @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding messages. "
+            + "Left to the client default when unset")
+    public Integer maxOutstanding;
 
     @Option(names = { "-p", "--max-outstanding-across-partitions" }, description = "Max number of outstanding "
-            + "messages across partitions")
-    public int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
+            + "messages across partitions. Left to the client default when unset")
+    public Integer maxPendingMessagesAcrossPartitions;
 
     @Option(names = { "-np", "--partitions" }, description = "Create partitioned topics with the given number "
             + "of partitions, set 0 to not try to create the topic")
@@ -447,14 +447,21 @@ static IMessageFormatter getMessageFormatter(String formatterClass) {
     }
 
     ProducerBuilder<byte[]> createProducerBuilder(PulsarClient client, int producerId) {
-        ProducerBuilder<byte[]> producerBuilder = client.newProducer() //
+        // Schema.BYTES rather than the no-argument newProducer(): the latter skips the pending-message
+        // defaults the client applies when its memory limit is disabled, which is how pulsar-perf runs
+        // unless --memory-limit is given.
+        ProducerBuilder<byte[]> producerBuilder = client.newProducer(Schema.BYTES) //
                 .sendTimeout(this.sendTimeout, TimeUnit.SECONDS) //
                 .compressionType(this.compression) //
-                .maxPendingMessages(this.maxOutstanding) //
                 .accessMode(this.producerAccessMode)
                 // enable round robin message routing if it is a partitioned topic
                 .messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
-        if (this.maxPendingMessagesAcrossPartitions > 0) {
+        // Only pass a limit the user actually asked for. Their unset value is 0, which the client reads
+        // as "no message-count limit", so passing it through would leave the producer unbounded.
+        if (this.maxOutstanding != null) {
+            producerBuilder.maxPendingMessages(this.maxOutstanding);
+        }
+        if (this.maxPendingMessagesAcrossPartitions != null) {
             producerBuilder.maxPendingMessagesAcrossPartitions(this.maxPendingMessagesAcrossPartitions);
         }
 

pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java

@@ -236,6 +236,49 @@ public void testMaxOutstanding() throws Exception {
         consumer.close();
     }
 
+    /**
+     * pulsar-perf runs with the client memory limit disabled unless {@code --memory-limit} is given, so
+     * the producer's only backpressure is its pending-message queue. Leaving the options unset has to
+     * leave the client's own defaults in place; passing their unset value of 0 through would read as an
+     * explicit "no message-count limit" and leave the producer unbounded, which is what exhausts direct
+     * memory against a slow broker on a non-partitioned topic.
+     */
+    @Test(timeOut = 20000)
+    public void testPendingMessageLimitsAreLeftToTheClientWhenUnset() throws Exception {
+        PerformanceProducer producer = new PerformanceProducer();
+        producer.topics = List.of(testTopic + UUID.randomUUID());
+        producer.serviceURL = pulsar.getBrokerServiceUrl();
+
+        @Cleanup
+        PulsarClient client = PerfClientUtils.createClientBuilderFromArguments(producer).build();
+        ProducerBuilderImpl<byte[]> builder =
+                (ProducerBuilderImpl<byte[]>) producer.createProducerBuilder(client, 0);
+
+        Assert.assertNull(producer.maxOutstanding);
+        Assert.assertNull(producer.maxPendingMessagesAcrossPartitions);
+        Assert.assertTrue(builder.getConf().getMaxPendingMessages() > 0);
+        Assert.assertTrue(builder.getConf().getMaxPendingMessagesAcrossPartitions() > 0);
+    }
+
+    /**
+     * An option that is given still wins over the client default, and either one can be given on its
+     * own — the across-partitions budget is allowed to be below the per-producer limit.
+     */
+    @Test(timeOut = 20000)
+    public void testGivenPendingMessageLimitsAreApplied() throws Exception {
+        PerformanceProducer producer = new PerformanceProducer();
+        producer.topics = List.of(testTopic + UUID.randomUUID());
+        producer.serviceURL = pulsar.getBrokerServiceUrl();
+        producer.maxPendingMessagesAcrossPartitions = 500;
+
+        @Cleanup
+        PulsarClient client = PerfClientUtils.createClientBuilderFromArguments(producer).build();
+        ProducerBuilderImpl<byte[]> builder =
+                (ProducerBuilderImpl<byte[]>) producer.createProducerBuilder(client, 0);
+
+        Assert.assertEquals(builder.getConf().getMaxPendingMessagesAcrossPartitions(), 500);
+    }
+
     @Test
     public void testRangeConvert() {
         PerformanceProducer.RangeConvert rangeConvert = new PerformanceProducer.RangeConvert();