You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/08/19 02:33:51 UTC
(pulsar) branch branch-4.2 updated: [fix][cli][branch-4.2] Don't pass unset pulsar-perf pending-message options to the producer (#26371)
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new e0d4fd3d880b [fix][cli][branch-4.2] Don't pass unset pulsar-perf pending-message options to the producer (#26371)
e0d4fd3d880b is described below
commit e0d4fd3d880b7b7f6fc9eb8be5b71d131644d19e
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Aug 19 05:33:44 2026 +0300
[fix][cli][branch-4.2] Don't pass unset pulsar-perf pending-message options to the producer (#26371)
---
.../impl/conf/ProducerConfigurationData.java | 12 +++++-
.../client/impl/ProducerBuilderImplTest.java | 17 ++++++++-
.../pulsar/testclient/PerformanceProducer.java | 25 ++++++++-----
.../pulsar/testclient/PerformanceProducerTest.java | 43 ++++++++++++++++++++++
4 files changed, 85 insertions(+), 12 deletions(-)
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 d324b6040ba0..1edfe72f6eb7 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 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: 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 738614398b12..31d1aa5e4405 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.any;
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 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, 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);
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 8860696321a3..25dd2f253c47 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.Objects.requireNonNull;
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.ProducerAccessMode;
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 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
}
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);
}
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 47815ddddf8b..4ed007527e65 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 class PerformanceProducerTest extends MockedPulsarServiceBaseTest {
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();