You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/16 11:35:10 UTC
(pulsar) 01/01: [improve][test] Support isolated v4 perf clients
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-perfopt-v4-isolated-clients
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 10ba198be38fee338fed627cb8f5aa2b5397d1a2
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Sep 16 12:34:13 2026 +0300
[improve][test] Support isolated v4 perf clients
---
.../pulsar/testclient/PerformanceConsumerBase.java | 22 +++++++++++++---
.../pulsar/testclient/PerformanceConsumerV4.java | 21 +++++++++++++++
.../pulsar/testclient/PerformanceProducerBase.java | 24 ++++++++++++-----
.../pulsar/testclient/PerformanceProducerV4.java | 23 +++++++++++++++++
.../pulsar/testclient/PulsarPerfTestToolTest.java | 30 +++++++++++++++++++---
.../profiling/AbstractPulsarProfilingTest.java | 6 +++++
.../profiling/PulsarProfilingConfig.java | 5 ++--
tests/performance/README.md | 5 ++++
tests/performance/pulsar-profiling.yaml | 3 +++
9 files changed, 125 insertions(+), 14 deletions(-)
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerBase.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerBase.java
index 0cc09cd2b6f2..e4616fc68399 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerBase.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerBase.java
@@ -220,6 +220,11 @@ public abstract class PerformanceConsumerBase<ClientT, ConsumerT, MessageT, TxnT
/** Close a client built by {@link #createClient()}; must tolerate a {@code null} argument. */
protected abstract void closeClient(ClientT client);
+ /** Number of clients used by this command. V4 can use one isolated client per consumer. */
+ protected int isolatedClientCount() {
+ return 0;
+ }
+
/** Subscribe one consumer to {@code topic} under {@code subscription}. */
protected abstract CompletableFuture<ConsumerT> subscribeAsync(ClientT client, String topic,
String subscription);
@@ -338,7 +343,16 @@ public abstract class PerformanceConsumerBase<ClientT, ConsumerT, MessageT, TxnT
this.testEndTime = startTime + (long) (this.testTime * 1e9);
this.mainThread = Thread.currentThread();
- this.client = createClient();
+ int isolatedClientCount = isolatedClientCount();
+ List<ClientT> clients = new ArrayList<>(isolatedClientCount > 0 ? isolatedClientCount : 1);
+ if (isolatedClientCount > 0) {
+ for (int i = 0; i < isolatedClientCount; i++) {
+ clients.add(createClient());
+ }
+ } else {
+ clients.add(createClient());
+ }
+ this.client = clients.get(0);
if (this.isEnableTransaction) {
this.transactionRef = new AtomicReference<>(openFirstTransaction(client));
@@ -350,6 +364,7 @@ public abstract class PerformanceConsumerBase<ClientT, ConsumerT, MessageT, TxnT
this.messageReceiveLimiter = new Semaphore(this.numMessagesPerTransaction);
List<CompletableFuture<ConsumerT>> futures = new ArrayList<>();
+ int consumerIndex = 0;
for (int i = 0; i < this.numTopics; i++) {
final TopicName topicName = TopicName.get(this.topics.get(i));
@@ -362,7 +377,8 @@ public abstract class PerformanceConsumerBase<ClientT, ConsumerT, MessageT, TxnT
for (int j = 0; j < this.numSubscriptions; j++) {
String subscription = this.subscriptions.get(j);
for (int k = 0; k < this.numConsumers; k++) {
- futures.add(subscribeAsync(client, topicName.toString(), subscription));
+ ClientT consumerClient = clients.get(consumerIndex++ % clients.size());
+ futures.add(subscribeAsync(consumerClient, topicName.toString(), subscription));
}
}
}
@@ -462,7 +478,7 @@ public abstract class PerformanceConsumerBase<ClientT, ConsumerT, MessageT, TxnT
}
// Stop driving the consumers before closing the client so receives do not race with close.
stopConsuming();
- closeClient(client);
+ clients.forEach(this::closeClient);
PerfClientUtils.removeAndRunShutdownHook(shutdownHookThread);
}
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerV4.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerV4.java
index 24d583b38924..1440c09b76e2 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerV4.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumerV4.java
@@ -64,6 +64,11 @@ public class PerformanceConsumerV4
@Option(names = { "-sp", "--subscription-position" }, description = "Subscription position")
private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Latest;
+ @Option(names = "--isolated-clients", description = "Create consumers on this many isolated v4 clients; "
+ + "cannot be combined with --num-listener-threads",
+ converter = PositiveNumberParameterConvert.class)
+ public int isolatedClients;
+
/** Receiver-queue depth samples, only allocated when {@code --auto-scaled-receiver-queue-size} is on. */
private Recorder qRecorder;
private Histogram qHistogram;
@@ -73,6 +78,22 @@ public class PerformanceConsumerV4
super("consume-v4");
}
+ @Override
+ public void validate() throws Exception {
+ super.validate();
+ if (isolatedClients > 0 && listenerThreads != 1) {
+ throw new IllegalArgumentException("--isolated-clients cannot be combined with --num-listener-threads");
+ }
+ if (isolatedClients > 0 && isEnableTransaction) {
+ throw new IllegalArgumentException("--isolated-clients cannot be used with transactions");
+ }
+ }
+
+ @Override
+ protected int isolatedClientCount() {
+ return isolatedClients;
+ }
+
@Override
protected void prepareRun() {
if (this.autoScaledReceiverQueueSize) {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerBase.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerBase.java
index f4d9c8b977ae..fbf425bfaa31 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerBase.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerBase.java
@@ -272,6 +272,16 @@ public abstract class PerformanceProducerBase<ClientT, ProducerT, TxnT> extends
/** Close a client built by {@link #createClient()}; must tolerate a {@code null} argument. */
protected abstract void closeClient(ClientT client);
+ /** Number of worker clients used by this command. V4 can use one client per producer. */
+ protected int workerCount() {
+ return this.numTestThreads;
+ }
+
+ /** Number of producers created by each worker. */
+ protected int producersPerWorker() {
+ return this.numProducers;
+ }
+
/**
* Create one producer on {@code topic}. {@code producerId} identifies the test thread and is
* only used to derive a unique producer name from {@code --producer-name}.
@@ -406,12 +416,13 @@ public abstract class PerformanceProducerBase<ClientT, ProducerT, TxnT> extends
}
}
- CountDownLatch doneLatch = new CountDownLatch(this.numTestThreads);
+ int workerCount = workerCount();
+ CountDownLatch doneLatch = new CountDownLatch(workerCount);
- final long numMessagesPerThread = this.numMessages / this.numTestThreads;
- final int msgRatePerThread = this.msgRate / this.numTestThreads;
+ final long numMessagesPerThread = this.numMessages / workerCount;
+ final int msgRatePerThread = this.msgRate / workerCount;
- for (int i = 0; i < this.numTestThreads; i++) {
+ for (int i = 0; i < workerCount; i++) {
final int threadIdx = i;
executor.submit(() -> {
log.info().attr("thread", threadIdx).log("Started performance test thread");
@@ -608,9 +619,10 @@ public abstract class PerformanceProducerBase<ClientT, ProducerT, TxnT> extends
for (int i = 0; i < this.numTopics; i++) {
String topic = this.topics.get(i);
- log.info().attr("adding", this.numProducers).attr("topic", topic).log("Adding publishers on topic");
+ log.info().attr("adding", producersPerWorker()).attr("topic", topic)
+ .log("Adding publishers on topic");
- for (int j = 0; j < this.numProducers; j++) {
+ for (int j = 0; j < producersPerWorker(); j++) {
futures.add(createProducerAsync(client, producerId, topic));
}
}
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerV4.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerV4.java
index b24e5efcc16f..1dfd26d89673 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerV4.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducerV4.java
@@ -54,10 +54,33 @@ public class PerformanceProducerV4
@Option(names = { "-am", "--access-mode" }, description = "Producer access mode")
public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
+ @Option(names = "--isolated-clients", description = "Create one isolated v4 client per producer; "
+ + "cannot be combined with --num-test-threads",
+ converter = PositiveNumberParameterConvert.class)
+ public int isolatedClients;
+
public PerformanceProducerV4() {
super("produce-v4");
}
+ @Override
+ public void validate() throws Exception {
+ super.validate();
+ if (isolatedClients > 0 && numTestThreads != 1) {
+ throw new IllegalArgumentException("--isolated-clients cannot be combined with --num-test-threads");
+ }
+ }
+
+ @Override
+ protected int workerCount() {
+ return isolatedClients > 0 ? isolatedClients : super.workerCount();
+ }
+
+ @Override
+ protected int producersPerWorker() {
+ return isolatedClients > 0 ? 1 : super.producersPerWorker();
+ }
+
@Override
protected PulsarClient createClient() throws PulsarClientException {
ClientBuilder clientBuilder = PerfClientUtils.createClientBuilderFromArguments(this)
diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PulsarPerfTestToolTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PulsarPerfTestToolTest.java
index 745f913392b2..279892adb181 100644
--- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PulsarPerfTestToolTest.java
+++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PulsarPerfTestToolTest.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.nio.file.Files;
import java.util.Arrays;
@@ -61,20 +62,26 @@ public class PulsarPerfTestToolTest {
});
}
- /** Each v4 command must offer the same flags as its V5 counterpart, so runs stay comparable. */
+ /** Each v4 command offers the V5 flags, plus v4-only controls where the clients differ. */
@Test
public void testV4CommandsOfferTheSameFlagsAsTheirV5Counterparts() throws Exception {
CommandLine commander = initCommander();
assertThat(optionNames(commander, "produce-v4"))
- .isEqualTo(optionNames(commander, "produce"));
+ .containsAll(optionNames(commander, "produce"))
+ .contains("--isolated-clients");
+ assertThat(optionNames(commander, "produce")).doesNotContain("--isolated-clients");
assertThat(optionNames(commander, "read-v4"))
.isEqualTo(optionNames(commander, "read"));
// consume and transaction have V5-only flags that the v4 commands must not advertise.
+ Set<String> consumeV4Options = optionNames(commander, "consume-v4");
+ consumeV4Options.remove("--isolated-clients");
assertThat(optionNames(commander, "consume"))
- .containsAll(optionNames(commander, "consume-v4"))
+ .containsAll(consumeV4Options)
.contains("--scalable-consumer-type");
assertThat(optionNames(commander, "consume-v4")).doesNotContain("--scalable-consumer-type");
+ assertThat(optionNames(commander, "consume-v4")).contains("--isolated-clients");
+ assertThat(optionNames(commander, "consume")).doesNotContain("--isolated-clients");
assertThat(optionNames(commander, "transaction"))
.containsAll(optionNames(commander, "transaction-v4"))
.contains("--scalable");
@@ -97,6 +104,23 @@ public class PulsarPerfTestToolTest {
assertThat(cmd.producerAccessMode).isEqualTo(ProducerAccessMode.Exclusive);
}
+ @Test
+ public void testIsolatedClientsAreV4OnlyAndCannotBeCombinedWithThreads() {
+ PerformanceProducerV4 producer = new PerformanceProducerV4();
+ new CommandLine(producer).parseArgs("--isolated-clients", "3", "topic");
+ assertThat(producer.isolatedClients).isEqualTo(3);
+
+ PerformanceProducerV4 conflictingProducer = new PerformanceProducerV4();
+ new CommandLine(conflictingProducer).parseArgs("--isolated-clients", "3", "-threads", "2", "topic");
+ assertThatThrownBy(conflictingProducer::validate)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("cannot be combined");
+
+ PerformanceConsumerV4 consumer = new PerformanceConsumerV4();
+ new CommandLine(consumer).parseArgs("--isolated-clients", "3", "topic");
+ assertThat(consumer.isolatedClients).isEqualTo(3);
+ }
+
/**
* The reports are emitted from the shared base classes, but must keep naming the concrete
* subcommand: {@code tests/integration/.../cli/PerfToolTest} matches stdout on
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/AbstractPulsarProfilingTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/AbstractPulsarProfilingTest.java
index 0f5d9f48eb61..675dee4ba329 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/AbstractPulsarProfilingTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/AbstractPulsarProfilingTest.java
@@ -164,6 +164,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
+ "-u pulsar://" + brokerHostname + ":6650 "
+ "-st Shared "
+ "-q 50000 "
+ + isolatedClientsOption(load.isolatedConsumers())
+ "-m " + load.numberOfMessages() + " -ml " + load.consumeMemoryLimit() + " "
+ "--histogram-file=/testoutput/consume" + commandSuffix
+ ".histogram.$(date +%s).hdr "
@@ -178,6 +179,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
+ "-au http://" + brokerHostname + ":8080 "
+ "-r " + load.produceRate() + " "
+ "-s " + load.messageSize() + " -db "
+ + isolatedClientsOption(load.isolatedProducers())
// maxOutstanding only applies to the v4 client; the v5 client accepts
// the flag for back-compat but ignores it
+ "-o " + load.maxOutstanding() + " "
@@ -187,6 +189,10 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
+ "2>&1 | tee /testoutput/produce" + commandSuffix + ".$(date +%s).txt");
}
+ private String isolatedClientsOption(int count) {
+ return commandSuffix.equals("-v4") && count > 0 ? "--isolated-clients " + count + " " : "";
+ }
+
/**
* Polls the given topic admin endpoints and the broker metrics every 10 seconds. The metrics
* are always collected; the topic endpoints may be empty when the domain has no admin stats.
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfig.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfig.java
index 230a3d7d5664..6df6e7c1d9d8 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfig.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfig.java
@@ -84,7 +84,7 @@ final class PulsarProfilingConfig {
Map.entry("isForceGCAllowWhenNoSpace", "true"),
Map.entry("diskUsageLwmThreshold", "0.75"),
Map.entry("diskCheckInterval", "60"))),
- new Load(20_000_000, "200M", "200M", Integer.MAX_VALUE, 128, 20_000, 10),
+ new Load(20_000_000, "200M", "200M", Integer.MAX_VALUE, 128, 20_000, 10, 0, 0),
new Output("build/pulsar-profiling"));
}
}
@@ -95,7 +95,8 @@ final class PulsarProfilingConfig {
}
record Load(long numberOfMessages, String produceMemoryLimit, String consumeMemoryLimit,
- int produceRate, int messageSize, int maxOutstanding, int statsIntervalSeconds) {
+ int produceRate, int messageSize, int maxOutstanding, int statsIntervalSeconds,
+ int isolatedProducers, int isolatedConsumers) {
}
record Output(String directory) {
diff --git a/tests/performance/README.md b/tests/performance/README.md
index 6510512f5955..cfd1aa46f90c 100644
--- a/tests/performance/README.md
+++ b/tests/performance/README.md
@@ -57,6 +57,11 @@ PULSAR_PROFILING_LOAD_NUMBER_OF_MESSAGES=1000000 \
./gradlew :tests:integration:profilingIntegrationTest --tests "*PulsarProfilingV4Test"
```
+For the v4 scenario, set `load.isolatedProducers` or `load.isolatedConsumers` to create that many
+independent v4 client instances. The corresponding `pulsar-perf` command receives
+`--isolated-clients`; v5 ignores these fields. The option is mutually exclusive with the regular
+producer test-thread option and with consumer listener-thread expansion.
+
## Inspecting recordings
Render the CPU, wall-clock, allocation and lock views with:
diff --git a/tests/performance/pulsar-profiling.yaml b/tests/performance/pulsar-profiling.yaml
index 0cd8b528e3cd..db1316d03a6c 100644
--- a/tests/performance/pulsar-profiling.yaml
+++ b/tests/performance/pulsar-profiling.yaml
@@ -38,5 +38,8 @@ load:
messageSize: 128
maxOutstanding: 20000
statsIntervalSeconds: 10
+ # Used only by produce-v4/consume-v4. A value of 500 creates up to 500 clients.
+ isolatedProducers: 0
+ isolatedConsumers: 0
output:
directory: build/pulsar-profiling