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 d324b6040ba03..1edfe72f6eb74 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: 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; } 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 738614398b12b..31d1aa5e44050 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 @@ -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 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); diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java index 8860696321a3e..25dd2f253c47b 100644 --- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java +++ b/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 createProducerBuilder(PulsarClient client, int producerId) { - ProducerBuilder 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 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); } diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java index 47815ddddf8b1..4ed007527e654 100644 --- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java +++ b/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 builder = + (ProducerBuilderImpl) 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 builder = + (ProducerBuilderImpl) producer.createProducerBuilder(client, 0); + + Assert.assertEquals(builder.getConf().getMaxPendingMessagesAcrossPartitions(), 500); + } + @Test public void testRangeConvert() { PerformanceProducer.RangeConvert rangeConvert = new PerformanceProducer.RangeConvert();