From 6f632a4454083a6e83d3eb1555a2ab158fff0f6d Mon Sep 17 00:00:00 2001 From: Alvaro <102966649+AlvaroStream@users.noreply.github.com> Date: Mon, 17 Aug 2026 09:38:44 +0200 Subject: [PATCH] [fix][cli] Restore the 64M default client memory limit in pulsar-perf Fixes #26340 Before #20663, pulsar-perf never called ClientBuilder#memoryLimit, so it inherited the client default of 64M from ClientConfigurationData. That PR added a --memory-limit option and applied it unconditionally, but the backing field has no initializer, so an unset option passes 0. Since MemoryLimitController#isMemoryLimited is memoryLimit > 0, 0 disables the limit entirely, and pulsar-perf silently went from bounded to unbounded client memory. With the limit disabled, a producer that outruns the brokers accumulates outbound buffers without backpressure until direct memory is exhausted, failing with OutOfDirectMemoryError instead of throttling. Default memoryLimit to the client's own default and extract that default into ClientConfigurationData.DEFAULT_MEMORY_LIMIT_BYTES so the two cannot drift. Passing --memory-limit 0 still disables the limit explicitly. --- .../impl/conf/ClientConfigurationData.java | 4 ++- .../testclient/PerformanceBaseArguments.java | 6 ++-- .../PerformanceBaseArgumentsTest.java | 33 +++++++++++++++++++ 3 files changed, 40 insertions(+), 3 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java index bbe9999045a5b..59ed56eb7cf62 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java @@ -61,6 +61,8 @@ public class ClientConfigurationData implements Serializable, Cloneable { private static final long serialVersionUID = 1L; + public static final long DEFAULT_MEMORY_LIMIT_BYTES = 64 * 1024 * 1024; + @Schema( name = "serviceUrl", requiredMode = Schema.RequiredMode.REQUIRED, @@ -433,7 +435,7 @@ public class ClientConfigurationData implements Serializable, Cloneable { description = "Limit of client memory usage (in byte). The 64M default can guarantee a high producer " + "throughput." ) - private long memoryLimitBytes = 64 * 1024 * 1024; + private long memoryLimitBytes = DEFAULT_MEMORY_LIMIT_BYTES; @Schema( name = "proxyServiceUrl", diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java index 3c4b831332281..3e5e0c90f0015 100644 --- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java @@ -19,6 +19,7 @@ package org.apache.pulsar.testclient; import static org.apache.commons.lang3.StringUtils.isBlank; +import static org.apache.pulsar.client.impl.conf.ClientConfigurationData.DEFAULT_MEMORY_LIMIT_BYTES; import org.apache.pulsar.cli.converters.picocli.ByteUnitToLongConverter; import org.apache.pulsar.client.api.ProxyProtocol; import picocli.CommandLine.Option; @@ -98,8 +99,9 @@ public abstract class PerformanceBaseArguments extends CmdBase{ public String deprecatedAuthPluginClassName; @Option(names = { "-ml", "--memory-limit", }, description = "Configure the Pulsar client memory limit " - + "(eg: 32M, 64M)", converter = ByteUnitToLongConverter.class) - public long memoryLimit; + + "(eg: 32M, 64M). Use 0 to disable the limit. Default: 64M", + converter = ByteUnitToLongConverter.class) + public long memoryLimit = DEFAULT_MEMORY_LIMIT_BYTES; public PerformanceBaseArguments(String cmdName) { super(cmdName); } diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java index 78cee5c22154b..3fe0b62fb5658 100644 --- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java +++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java @@ -19,6 +19,7 @@ package org.apache.pulsar.testclient; import static org.apache.pulsar.client.api.ProxyProtocol.SNI; +import static org.apache.pulsar.client.impl.conf.ClientConfigurationData.DEFAULT_MEMORY_LIMIT_BYTES; import static org.testng.Assert.assertEquals; import static org.testng.Assert.fail; import java.io.File; @@ -249,6 +250,38 @@ public void run() throws Exception { // Act baseArgument.parseCLI(); + // Assert + assertEquals(baseArgument.memoryLimit, DEFAULT_MEMORY_LIMIT_BYTES); + } + } + + @Test + public void testMemoryLimitCanBeDisabled() throws Exception { + for (String cmd : List.of( + "pulsar-perf read", + "pulsar-perf produce", + "pulsar-perf consume", + "pulsar-perf transaction" + )) { + // Arrange + final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments("") { + @Override + public void run() throws Exception { + + } + + }; + String confFile = "./src/test/resources/perf_client1.conf"; + Properties prop = new Properties(System.getProperties()); + try (FileInputStream fis = new FileInputStream(confFile)) { + prop.load(fis); + } + baseArgument.getCommander().setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop)); + baseArgument.parse(new String[]{"-ml", "0"}); + + // Act + baseArgument.parseCLI(); + // Assert assertEquals(baseArgument.memoryLimit, 0L); }