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/17 02:27:15 UTC
(pulsar) branch lh-improve-read-completion created (now bf5615609d1e)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch lh-improve-read-completion
in repository https://gitbox.apache.org/repos/asf/pulsar.git
at bf5615609d1e [improve][broker] Avoid redundant read-completion executor handoffs
This branch includes the following new commits:
new bf5615609d1e [improve][broker] Avoid redundant read-completion executor handoffs
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(pulsar) 01/01: [improve][broker] Avoid redundant read-completion executor handoffs
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-improve-read-completion
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bf5615609d1e8ea5d4853e7d98720d2052514a45
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Thu Sep 17 05:14:27 2026 +0300
[improve][broker] Avoid redundant read-completion executor handoffs
Allow self-dispatching read callbacks to complete on the current thread while retaining the Exclusive/Failover dispatcher executor boundary. Add callback affinity and failure coverage, a handoff benchmark, and reproducible YAML profiling scenarios.
Assisted-by: Codex
---
.../apache/bookkeeper/mledger/AsyncCallbacks.java | 9 +
.../bookkeeper/mledger/impl/OpReadEntry.java | 4 +
.../mledger/util/ManagedLedgerUtils.java | 15 ++
.../mledger/impl/ReadCompletionAffinityTest.java | 200 ++++++++++++++++++
.../impl/ReadCompletionHandoffBenchmark.java | 80 ++++++++
.../PersistentDispatcherSingleActiveConsumer.java | 4 +-
.../profiling/AbstractPulsarProfilingTest.java | 37 +++-
.../profiling/PulsarProfilingConfig.java | 99 +++++++--
.../profiling/PulsarProfilingConfigTest.java | 226 +++++++++++++++++++++
.../profiling/PulsarProfilingV4Test.java | 12 ++
tests/performance/README.md | 68 ++++++-
tests/performance/read-completion-isolation.md | 123 +++++++++++
.../{ => scenarios}/pulsar-profiling.yaml | 12 ++
.../read-completion-isolation-256k-128k.yaml | 25 +++
.../read-completion-isolation-64k-32k.yaml | 25 +++
.../read-completion-isolation-failover.yaml | 23 +++
.../read-completion-isolation-shared.yaml | 23 +++
.../read-completion-isolation.yaml} | 36 ++--
18 files changed, 979 insertions(+), 42 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/AsyncCallbacks.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/AsyncCallbacks.java
index 70db427afce4..59e276aafdd8 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/AsyncCallbacks.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/AsyncCallbacks.java
@@ -83,6 +83,15 @@ public interface AsyncCallbacks {
}
interface ReadEntriesCallback {
+ /**
+ * Whether a successful read may invoke this callback on the completing thread, avoiding a handoff to the
+ * managed-ledger executor. The callback must not block and must bound any recursive reads it initiates.
+ * Callbacks that dispatch their own continuation to an executor can opt in without changing its affinity.
+ */
+ default boolean canExecuteOnAnyThread() {
+ return false;
+ }
+
void readEntriesComplete(List<Entry> entries, Object ctx);
void readEntriesFailed(ManagedLedgerException exception, Object ctx);
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
index 153db59ab7f4..c82781d93fc5 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
@@ -286,6 +286,10 @@ class OpReadEntry implements ReadEntriesCallback {
}
private void complete(Object ctx) {
+ if (callback.canExecuteOnAnyThread()) {
+ completeNow(ctx);
+ return;
+ }
ThreadBoundExecutor executor = cursor.ledger.getExecutor();
// Run inline on the ledger thread to skip the queue hop. A fully cached read completes synchronously and
// callers such as OpScan and the replicator issue their next read from this callback, so the nesting is
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerUtils.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerUtils.java
index 8ce8a2eef69d..14869ac08125 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerUtils.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/ManagedLedgerUtils.java
@@ -84,8 +84,23 @@ public class ManagedLedgerUtils {
public static CompletableFuture<List<Entry>> readEntriesWithSkipOrWait(
ManagedCursor cursor, int maxEntries, long maxSizeBytes, Position maxPosition,
@Nullable Predicate<Position> skipCondition) {
+ return readEntriesWithSkipOrWait(cursor, maxEntries, maxSizeBytes, maxPosition, skipCondition, false);
+ }
+
+ /**
+ * Reads entries with optional completion on the reading thread. When enabled, callers must dispatch continuations
+ * to their own executor if they require thread affinity or an asynchronous boundary between reads.
+ */
+ public static CompletableFuture<List<Entry>> readEntriesWithSkipOrWait(
+ ManagedCursor cursor, int maxEntries, long maxSizeBytes, Position maxPosition,
+ @Nullable Predicate<Position> skipCondition, boolean callbackCanExecuteOnAnyThread) {
final var future = new CompletableFuture<List<Entry>>();
cursor.asyncReadEntriesWithSkipOrWait(maxEntries, maxSizeBytes, new AsyncCallbacks.ReadEntriesCallback() {
+ @Override
+ public boolean canExecuteOnAnyThread() {
+ return callbackCanExecuteOnAnyThread;
+ }
+
@Override
public void readEntriesComplete(List<Entry> entries, Object ctx) {
future.complete(entries);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ReadCompletionAffinityTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ReadCompletionAffinityTest.java
new file mode 100644
index 000000000000..cc015f462935
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ReadCompletionAffinityTest.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import static org.apache.bookkeeper.mledger.util.ManagedLedgerTestUtil.rawEntryConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.util.ManagedLedgerUtils;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ReadCompletionAffinityTest extends MockedBookKeeperTestCase {
+ @DataProvider
+ public Object[][] inlineCompletion() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "inlineCompletion")
+ public void testCacheHitCompletionAffinity(boolean inline) throws Exception {
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("completion-affinity-" + inline,
+ rawEntryConfig());
+ CountDownLatch releaseWorker = new CountDownLatch(1);
+ try {
+ ManagedCursor cursor = ledger.openCursor("cursor");
+ ledger.addEntry(new byte[] {1});
+ assertTrue(ledger.entryCache.getSize() > 0, "The test requires a cache hit");
+ Thread worker = blockWorker(ledger, releaseWorker);
+ Thread caller = Thread.currentThread();
+ CompletableFuture<Thread> completed = new CompletableFuture<>();
+ cursor.asyncReadEntries(1, new ReadEntriesCallback() {
+ @Override
+ public boolean canExecuteOnAnyThread() {
+ return inline;
+ }
+
+ @Override
+ public void readEntriesComplete(List<Entry> entries, Object ctx) {
+ entries.forEach(Entry::release);
+ completed.complete(Thread.currentThread());
+ }
+
+ @Override
+ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ completed.completeExceptionally(exception);
+ }
+ }, null, PositionFactory.LATEST);
+ // A blocked worker makes the queue boundary deterministic, rather than depending on task timing.
+ assertEquals(completed.isDone(), inline);
+ releaseWorker.countDown();
+ assertSame(completed.get(10, TimeUnit.SECONDS), inline ? caller : worker);
+ } finally {
+ releaseWorker.countDown();
+ ledger.close();
+ }
+ }
+
+ @Test
+ public void testInlineFutureKeepsDispatcherBoundary() throws Exception {
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("completion-dispatcher-boundary",
+ rawEntryConfig());
+ CountDownLatch releaseWorker = new CountDownLatch(1);
+ ExecutorService dispatcher = Executors.newSingleThreadExecutor();
+ try {
+ ManagedCursor cursor = ledger.openCursor("cursor");
+ ledger.addEntry(new byte[] {1});
+ assertTrue(ledger.entryCache.getSize() > 0, "The test requires a cache hit");
+ Thread worker = blockWorker(ledger, releaseWorker);
+ Thread caller = Thread.currentThread();
+ CompletableFuture<List<Entry>> read = ManagedLedgerUtils.readEntriesWithSkipOrWait(
+ cursor, 1, Long.MAX_VALUE, PositionFactory.LATEST, null, true);
+ CompletableFuture<Thread> continued = read.thenApplyAsync(entries -> {
+ entries.forEach(Entry::release);
+ return Thread.currentThread();
+ }, dispatcher);
+ assertTrue(read.isDone(), "A cache hit must not wait behind unrelated ledger work");
+ Thread continuationThread = continued.get(10, TimeUnit.SECONDS);
+ assertNotSame(continuationThread, caller);
+ assertNotSame(continuationThread, worker);
+ } finally {
+ releaseWorker.countDown();
+ ledger.close();
+ dispatcher.shutdownNow();
+ assertTrue(dispatcher.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+
+ private static Thread blockWorker(ManagedLedgerImpl ledger, CountDownLatch releaseWorker) throws Exception {
+ CompletableFuture<Thread> started = new CompletableFuture<>();
+ ledger.getExecutor().execute(() -> {
+ started.complete(Thread.currentThread());
+ try {
+ releaseWorker.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ return started.get(10, TimeUnit.SECONDS);
+ }
+
+ @Test(dataProvider = "inlineCompletion")
+ public void testCacheMissAcrossLedgers(boolean inline) throws Exception {
+ ManagedLedgerConfig config = rawEntryConfig().setMaxEntriesPerLedger(2);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("completion-rollover-" + inline, config);
+ AtomicInteger storageReads = new AtomicInteger();
+ try {
+ ManagedCursor cursor = ledger.openCursor("cursor");
+ for (int i = 0; i < 6; i++) {
+ ledger.addEntry(new byte[] {(byte) i});
+ }
+ ledger.entryCache.clear();
+ bkc.setReadHandleInterceptor((ledgerId, first, last, entries) -> {
+ storageReads.incrementAndGet();
+ return CompletableFuture.completedFuture(entries);
+ });
+ List<Entry> entries = ManagedLedgerUtils.readEntriesWithSkipOrWait(
+ cursor, 6, Long.MAX_VALUE, PositionFactory.LATEST, null, inline).get(10, TimeUnit.SECONDS);
+ try {
+ assertThat(entries).hasSize(6);
+ for (int i = 0; i < entries.size(); i++) {
+ assertThat(entries.get(i).getData()).containsExactly((byte) i);
+ }
+ assertThat(entries.stream().map(entry -> entry.getPosition().getLedgerId()).distinct().count())
+ .isEqualTo(3);
+ assertThat(storageReads.get()).isGreaterThanOrEqualTo(3);
+ } finally {
+ entries.forEach(Entry::release);
+ }
+ } finally {
+ bkc.setReadHandleInterceptor(null);
+ ledger.close();
+ }
+ }
+
+ @Test(dataProvider = "inlineCompletion")
+ public void testStorageFailureAndRetry(boolean inline) throws Exception {
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("completion-failure-" + inline,
+ rawEntryConfig());
+ try {
+ ManagedCursor cursor = ledger.openCursor("cursor");
+ ledger.addEntry(new byte[] {1});
+ ledger.entryCache.clear();
+ bkc.setReadHandleInterceptor((ledgerId, first, last, entries) -> {
+ entries.close();
+ return CompletableFuture.failedFuture(BKException.create(BKException.Code.ReadException));
+ });
+ var failed = ManagedLedgerUtils.readEntriesWithSkipOrWait(
+ cursor, 1, Long.MAX_VALUE, PositionFactory.LATEST, null, inline);
+ assertThatThrownBy(() -> failed.get(10, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(ManagedLedgerException.class);
+ bkc.setReadHandleInterceptor(null);
+ List<Entry> entries = ManagedLedgerUtils.readEntriesWithSkipOrWait(
+ cursor, 1, Long.MAX_VALUE, PositionFactory.LATEST, null, inline).get(10, TimeUnit.SECONDS);
+ try {
+ assertThat(entries).hasSize(1);
+ assertThat(entries.get(0).getData()).containsExactly((byte) 1);
+ } finally {
+ entries.forEach(Entry::release);
+ }
+ } finally {
+ bkc.setReadHandleInterceptor(null);
+ ledger.close();
+ }
+ }
+}
diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ReadCompletionHandoffBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ReadCompletionHandoffBenchmark.java
new file mode 100644
index 000000000000..2789ac63e2d0
--- /dev/null
+++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/ReadCompletionHandoffBenchmark.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.common.util.SingleThreadExecutor;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+/** Measures the future-adapter handoff using real executors, without ledger/cache work or producer queue pressure. */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class ReadCompletionHandoffBenchmark {
+ @Param({"false", "true"})
+ public boolean inline;
+
+ private SingleThreadExecutor ledger;
+ private SingleThreadExecutor dispatcher;
+
+ @Setup
+ public void setup() {
+ ledger = new SingleThreadExecutor(new DefaultThreadFactory("ledger-benchmark"));
+ dispatcher = new SingleThreadExecutor(new DefaultThreadFactory("dispatcher-benchmark"));
+ }
+
+ @Benchmark
+ public int completeRead() {
+ CompletableFuture<Integer> completed = new CompletableFuture<>();
+ dispatcher.execute(() -> {
+ CompletableFuture<Integer> read = new CompletableFuture<>();
+ if (inline) {
+ read.complete(1);
+ } else {
+ ledger.executeOrRun(() -> read.complete(1));
+ }
+ read.thenAcceptAsync(completed::complete, dispatcher);
+ });
+ return completed.join();
+ }
+
+ @TearDown
+ public void tearDown() throws InterruptedException {
+ ledger.shutdown();
+ dispatcher.shutdown();
+ ledger.awaitTermination(10, TimeUnit.SECONDS);
+ dispatcher.awaitTermination(10, TimeUnit.SECONDS);
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 8c2b1d820520..80f7e39492ee 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -397,8 +397,10 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
entriesFuture = CompactedTopicUtils.asyncReadCompactedEntries(topic.getTopicCompactionService(),
cursor, messagesToRead, bytesToRead, topic.getMaxReadPosition(), readFromEarliest, true);
} else {
+ // The continuation below already dispatches to our executor. A cache hit must not wait
+ // behind unrelated publishing work on the managed-ledger executor first.
entriesFuture = readEntriesWithSkipOrWait(cursor, messagesToRead, bytesToRead,
- topic.getMaxReadPosition(), null);
+ topic.getMaxReadPosition(), null, true);
}
entriesFuture.whenCompleteAsync((entries, e) -> {
if (e == null) {
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 675dee4ba329..c2982c821e93 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
@@ -19,6 +19,7 @@
package org.apache.pulsar.tests.integration.profiling;
import static org.assertj.core.api.SoftAssertions.assertSoftly;
+import com.github.dockerjava.api.model.Capability;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
@@ -33,6 +34,7 @@ import lombok.CustomLog;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.tests.ManualTestUtil;
import org.apache.pulsar.tests.integration.containers.PulsarContainer;
import org.apache.pulsar.tests.integration.suites.PulsarTestSuite;
@@ -162,8 +164,10 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
"bash", "-c", "set -o pipefail; echo $$ > /tmp/command.pid; "
+ "/pulsar/bin/pulsar-perf consume" + commandSuffix + " " + topicName + " "
+ "-u pulsar://" + brokerHostname + ":6650 "
- + "-st Shared "
- + "-q 50000 "
+ + "-st " + load.subscriptionType() + " "
+ + "-q " + load.receiverQueueSize() + " "
+ + "--num-consumers " + load.consumerCount() + " "
+ + "--num-io-threads " + load.consumerIoThreads() + " "
+ isolatedClientsOption(load.isolatedConsumers())
+ "-m " + load.numberOfMessages() + " -ml " + load.consumeMemoryLimit() + " "
+ "--histogram-file=/testoutput/consume" + commandSuffix
@@ -178,7 +182,11 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
+ "-u pulsar://" + brokerHostname + ":6650 "
+ "-au http://" + brokerHostname + ":8080 "
+ "-r " + load.produceRate() + " "
- + "-s " + load.messageSize() + " -db "
+ + "-s " + load.messageSize() + " "
+ + (load.batchingEnabled() ? "" : "-db ")
+ + "--num-producers " + load.producerCount() + " "
+ + "--num-io-threads " + load.producerIoThreads() + " "
+ + "--max-connections 1 "
+ isolatedClientsOption(load.isolatedProducers())
// maxOutstanding only applies to the v4 client; the v5 client accepts
// the flag for back-compat but ignores it
@@ -193,6 +201,19 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
return commandSuffix.equals("-v4") && count > 0 ? "--isolated-clients " + count + " " : "";
}
+ void enableProfiling(String options, String role) {
+ if (options == null || options.isBlank()) {
+ return;
+ }
+ // Permit native CPU sampling, as for the profiled broker containers.
+ withCreateContainerCmdModifier(cmd -> cmd.getHostConfig()
+ .withCapAdd(Capability.PERFMON)
+ .withSecurityOpts(List.of("seccomp=unconfined")));
+ withEnv("PULSAR_EXTRA_OPTS", "-XX:+UnlockDiagnosticVMOptions -XX:+DebugNonSafepoints "
+ + "-agentpath:/opt/async-profiler/lib/libasyncProfiler.so=start," + options
+ + ",file=/testoutput/client-" + role + "-%t-%p.jfr");
+ }
+
/**
* 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.
@@ -270,6 +291,8 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
// This matters only on Linux
try {
Files.setPosixFilePermissions(testOutputDir.toPath(), PosixFilePermissions.fromString("rwxrwxrwx"));
+ ObjectMapperFactory.getYamlMapper().getObjectMapper().writeValue(
+ new File(testOutputDir, "resolved-config.yaml"), profilingConfig);
} catch (IOException e) {
throw new UncheckedIOException("Cannot change access to test output directory", e);
}
@@ -352,6 +375,8 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
commandSuffix, load);
perfConsume = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "perf-consume", "-Xmx1g",
commandSuffix, load);
+ perfProduce.enableProfiling(profilingConfig.profiling().producerOptions(), "producer");
+ perfConsume.enableProfiling(profilingConfig.profiling().consumerOptions(), "consumer");
printStats = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "print-stats", "-Xmx1g",
commandSuffix, load);
specBuilder.externalServices(Map.of(
@@ -374,6 +399,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
*/
protected void runPulsarPerfBenchmark() throws Exception {
String topicName = generateTopicName("profiletest", getTopicDomain());
+ prepareTopic(topicName);
CompletableFuture<Long> consumeFuture = perfConsume.consume(topicName);
Thread.sleep(1000);
CompletableFuture<Long> produceFuture = perfProduce.produce(topicName);
@@ -381,7 +407,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
printStats.stats(getTopicStatsEndpoints(topicName));
// pulsar-perf is sized to finish inside this window, so running out of it is a failure.
FutureUtil.waitForAll(List.of(consumeFuture, produceFuture))
- .orTimeout(3, TimeUnit.MINUTES)
+ .orTimeout(profilingConfig.load().timeoutSeconds(), TimeUnit.SECONDS)
.exceptionally(t -> {
log.error().exception(t).log("Failed to run pulsar-perf");
throw FutureUtil.wrapToCompletionException(t);
@@ -392,4 +418,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
softly.assertThat(produceFuture).as("produce should have completed successfully").isCompletedWithValue(0L);
});
}
+
+ protected void prepareTopic(String topicName) throws Exception {
+ }
}
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 6df6e7c1d9d8..4249b0a26360 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
@@ -23,7 +23,10 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;
import java.nio.file.Path;
+import java.util.LinkedHashSet;
import java.util.Map;
+import java.util.Set;
+import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.util.ObjectMapperFactory;
/** Configuration for the profiling scenario harness. */
@@ -31,7 +34,7 @@ final class PulsarProfilingConfig {
static final String CONFIG_ENV = "PULSAR_PROFILING_CONFIG";
static final String ENV_PREFIX = "PULSAR_PROFILING_";
- record Config(Cluster cluster, Load load, Output output) {
+ record Config(Cluster cluster, Load load, Profiling profiling, Output output) {
static Config read() {
String configFile = System.getenv(CONFIG_ENV);
return read(configFile == null || configFile.isBlank() ? null : Path.of(configFile),
@@ -42,11 +45,7 @@ final class PulsarProfilingConfig {
ObjectMapper mapper = ObjectMapperFactory.getYamlMapper().getObjectMapper();
ObjectNode root = mapper.valueToTree(defaults());
if (configFile != null) {
- try {
- merge(root, mapper.readTree(configFile.toFile()));
- } catch (IOException e) {
- throw new IllegalArgumentException("Cannot read profiling config " + configFile, e);
- }
+ mergeFile(root, configFile, mapper, new LinkedHashSet<>());
}
applyEnvironmentOverrides(root, environment);
try {
@@ -84,7 +83,9 @@ 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, 0, 0),
+ new Load(20_000_000, "200M", "200M", Integer.MAX_VALUE, 128, 20_000, 10, 0, 0,
+ 1, 1, 1, 1, SubscriptionType.Shared, 50_000, 180, false),
+ new Profiling("", ""),
new Output("build/pulsar-profiling"));
}
}
@@ -96,22 +97,90 @@ final class PulsarProfilingConfig {
record Load(long numberOfMessages, String produceMemoryLimit, String consumeMemoryLimit,
int produceRate, int messageSize, int maxOutstanding, int statsIntervalSeconds,
- int isolatedProducers, int isolatedConsumers) {
+ int isolatedProducers, int isolatedConsumers, int producerCount, int consumerCount,
+ int producerIoThreads, int consumerIoThreads, SubscriptionType subscriptionType,
+ int receiverQueueSize, int timeoutSeconds, boolean batchingEnabled) {
+ Load {
+ if (producerCount < 1 || consumerCount < 1 || producerIoThreads < 1 || consumerIoThreads < 1
+ || isolatedProducers < 0 || isolatedConsumers < 0 || receiverQueueSize < 1 || timeoutSeconds < 1
+ || numberOfMessages < 1 || produceRate < 1 || subscriptionType == null) {
+ throw new IllegalArgumentException("Profiling counts, rate, queue size and timeout must be positive");
+ }
+ // pulsar-perf divides the message count and rate between workers using integer division.
+ int workers = Math.max(1, isolatedProducers);
+ if (isolatedProducers > producerCount || isolatedConsumers > consumerCount
+ || numberOfMessages % workers != 0 || produceRate < workers) {
+ throw new IllegalArgumentException("Isolated clients require at least one producer/consumer "
+ + "per client, a message count divisible by producer clients, and a rate >= producer clients");
+ }
+ if (subscriptionType == SubscriptionType.Exclusive && consumerCount != 1) {
+ throw new IllegalArgumentException("Exclusive subscriptions require exactly one consumer");
+ }
+ }
}
record Output(String directory) {
}
- private static void merge(ObjectNode target, JsonNode source) {
- if (source == null || !source.isObject()) {
- return;
+ /** Empty options disable profiling for that client process. */
+ record Profiling(String producerOptions, String consumerOptions) {
+ }
+
+ private static void mergeFile(ObjectNode target, Path file, ObjectMapper mapper, Set<Path> activeFiles) {
+ try {
+ Path path = file.toRealPath();
+ if (!activeFiles.add(path)) {
+ throw new IllegalArgumentException("Profiling config inheritance cycle: "
+ + activeFiles + " -> " + path);
+ }
+ try {
+ JsonNode source = mapper.readTree(path.toFile());
+ if (!(source instanceof ObjectNode object)) {
+ throw new IllegalArgumentException("Profiling config must be a YAML mapping: " + path);
+ }
+ JsonNode parents = object.remove("extends");
+ if (parents != null) {
+ if (parents.isTextual()) {
+ mergeParent(target, parents, path, mapper, activeFiles);
+ } else if (parents.isArray()) {
+ for (JsonNode parent : parents) {
+ mergeParent(target, parent, path, mapper, activeFiles);
+ }
+ } else {
+ throw new IllegalArgumentException(
+ "Profiling config 'extends' must be a path or list of paths: " + path);
+ }
+ }
+ // Apply directly to the accumulated config so inherited nulls also remove earlier values/defaults.
+ merge(target, object);
+ } finally {
+ activeFiles.remove(path);
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot read profiling config " + file, e);
+ }
+ }
+
+ private static void mergeParent(ObjectNode target, JsonNode parent, Path file,
+ ObjectMapper mapper, Set<Path> activeFiles) {
+ if (!parent.isTextual() || parent.textValue().isBlank()) {
+ throw new IllegalArgumentException("Profiling config 'extends' entries must be non-empty paths: " + file);
}
+ mergeFile(target, file.getParent().resolve(parent.textValue()), mapper, activeFiles);
+ }
+
+ private static void merge(ObjectNode target, ObjectNode source) {
source.properties().forEach(entry -> {
- JsonNode current = target.get(entry.getKey());
- if (current != null && current.isObject() && entry.getValue().isObject()) {
- merge((ObjectNode) current, entry.getValue());
+ String key = entry.getKey();
+ JsonNode value = entry.getValue();
+ if (value.isNull()) {
+ target.remove(key);
+ } else if (value instanceof ObjectNode object) {
+ JsonNode current = target.get(key);
+ ObjectNode child = current instanceof ObjectNode ? (ObjectNode) current : target.putObject(key);
+ merge(child, object);
} else {
- target.set(entry.getKey(), entry.getValue());
+ target.set(key, value);
}
});
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfigTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfigTest.java
index 472d8f6a4894..5515b0cfa449 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfigTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfigTest.java
@@ -19,9 +19,14 @@
package org.apache.pulsar.tests.integration.profiling;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Comparator;
import java.util.Map;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class PulsarProfilingConfigTest {
@@ -42,4 +47,225 @@ public class PulsarProfilingConfigTest {
Files.deleteIfExists(file);
}
}
+
+ @Test
+ public void configuresIsolatedProducersAndExclusiveConsumer() {
+ var config = PulsarProfilingConfig.Config.read(null, Map.of(
+ "PULSAR_PROFILING_LOAD_PRODUCER_COUNT", "500",
+ "PULSAR_PROFILING_LOAD_ISOLATED_PRODUCERS", "500",
+ "PULSAR_PROFILING_LOAD_PRODUCER_IO_THREADS", "8",
+ "PULSAR_PROFILING_LOAD_SUBSCRIPTION_TYPE", "Exclusive",
+ "PULSAR_PROFILING_LOAD_TIMEOUT_SECONDS", "480"));
+ assertThat(config.load().producerCount()).isEqualTo(500);
+ assertThat(config.load().isolatedProducers()).isEqualTo(500);
+ assertThat(config.load().producerIoThreads()).isEqualTo(8);
+ assertThat(config.load().subscriptionType()).isEqualTo(SubscriptionType.Exclusive);
+ assertThat(config.load().consumerCount()).isEqualTo(1);
+ assertThat(config.load().timeoutSeconds()).isEqualTo(480);
+ }
+
+ @Test
+ public void inheritsIndependentClientProfilingAndBatchingOptions() throws Exception {
+ try (ScenarioFiles files = new ScenarioFiles()) {
+ files.write("clients.yaml", """
+ profiling:
+ producerOptions: event=cpu,interval=10ms
+ consumerOptions: event=cpu,lock=0
+ load:
+ batchingEnabled: true
+ """);
+ Path child = files.write("scenario.yaml", """
+ extends: clients.yaml
+ profiling:
+ consumerOptions: ~
+ """);
+ var config = PulsarProfilingConfig.Config.read(child, Map.of(
+ "PULSAR_PROFILING_PROFILING_PRODUCER_OPTIONS", "event=cpu,interval=20ms",
+ "PULSAR_PROFILING_LOAD_BATCHING_ENABLED", "false"));
+ assertThat(config.profiling().producerOptions()).isEqualTo("event=cpu,interval=20ms");
+ assertThat(config.profiling().consumerOptions()).isNull();
+ assertThat(config.load().batchingEnabled()).isFalse();
+ assertThat(PulsarProfilingConfig.Config.defaults().profiling().producerOptions()).isEmpty();
+ assertThat(PulsarProfilingConfig.Config.defaults().profiling().consumerOptions()).isEmpty();
+ }
+ }
+
+ @Test
+ public void rejectsWorkloadsThatCannotFinish() {
+ assertThatThrownBy(() -> PulsarProfilingConfig.Config.read(null,
+ Map.of("PULSAR_PROFILING_LOAD_ISOLATED_PRODUCERS", "500")))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> PulsarProfilingConfig.Config.read(null, Map.of(
+ "PULSAR_PROFILING_LOAD_PRODUCER_COUNT", "500",
+ "PULSAR_PROFILING_LOAD_ISOLATED_PRODUCERS", "500",
+ "PULSAR_PROFILING_LOAD_NUMBER_OF_MESSAGES", "501")))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> PulsarProfilingConfig.Config.read(null, Map.of(
+ "PULSAR_PROFILING_LOAD_SUBSCRIPTION_TYPE", "Exclusive",
+ "PULSAR_PROFILING_LOAD_CONSUMER_COUNT", "2")))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ public void mergesRecursiveParentsInOrderBeforeChildAndEnvironment() throws Exception {
+ try (ScenarioFiles files = new ScenarioFiles()) {
+ files.write("common.yaml", """
+ load:
+ messageSize: 256
+ produceRate: 1000
+ cluster:
+ numBookies: 5
+ """);
+ files.write("parents/first.yaml", """
+ extends: ../common.yaml
+ load:
+ messageSize: 512
+ producerCount: 3
+ """);
+ files.write("parents/second.yaml", """
+ load:
+ messageSize: 1024
+ produceRate: 2000
+ """);
+ Path child = files.write("scenario.yaml", """
+ extends: [parents/first.yaml, parents/second.yaml]
+ load:
+ produceRate: 3000
+ """);
+ var config = PulsarProfilingConfig.Config.read(child, Map.of(
+ "PULSAR_PROFILING_CLUSTER_NUM_BOOKIES", "7"));
+ assertThat(config.load().messageSize()).isEqualTo(1024);
+ assertThat(config.load().produceRate()).isEqualTo(3000);
+ assertThat(config.load().producerCount()).isEqualTo(3);
+ assertThat(config.cluster().numBookies()).isEqualTo(7);
+ assertThat(config.load().consumerCount()).isEqualTo(1);
+ }
+ }
+
+ @Test
+ public void removesInheritedAndDefaultEntriesIncludingThroughRecursiveParents() throws Exception {
+ try (ScenarioFiles files = new ScenarioFiles()) {
+ files.write("base.yaml", """
+ cluster:
+ brokerEnvs:
+ customSetting: enabled
+ keptSetting: kept
+ restoredSetting: old
+ """);
+ files.write("removals.yaml", """
+ cluster:
+ brokerEnvs:
+ customSetting: ~
+ preciseDispatcherFlowControl: null
+ restoredSetting: ~
+ absentSetting: ~
+ bookkeeperEnvs: ~
+ """);
+ files.write("parent.yaml", "extends: removals.yaml\n");
+ Path child = files.write("scenario.yaml", """
+ extends: [base.yaml, parent.yaml]
+ cluster:
+ brokerEnvs:
+ restoredSetting: new
+ bookkeeperEnvs:
+ journalSyncData: "true"
+ absentSetting: ~
+ """);
+ var config = PulsarProfilingConfig.Config.read(child, Map.of());
+ assertThat(config.cluster().brokerEnvs())
+ .doesNotContainKeys("customSetting", "preciseDispatcherFlowControl", "absentSetting")
+ .containsEntry("keptSetting", "kept")
+ .containsEntry("restoredSetting", "new")
+ .containsEntry("managedLedgerDefaultEnsembleSize", "1");
+ assertThat(config.cluster().bookkeeperEnvs()).containsExactlyEntriesOf(Map.of("journalSyncData", "true"));
+ }
+ }
+
+ @Test
+ public void allowsSharedAncestorsWithoutTreatingThemAsCycles() throws Exception {
+ try (ScenarioFiles files = new ScenarioFiles()) {
+ files.write("common.yaml", "load:\n messageSize: 256\n");
+ files.write("first.yaml", "extends: common.yaml\nload:\n messageSize: 512\n");
+ files.write("second.yaml", "extends: common.yaml\n");
+ Path child = files.write("scenario.yaml", "extends: [first.yaml, second.yaml]\n");
+ var config = PulsarProfilingConfig.Config.read(child, Map.of());
+ // The second parent's ancestry is applied too, in declaration order.
+ assertThat(config.load().messageSize()).isEqualTo(256);
+ }
+ }
+
+ @Test
+ public void rejectsInheritanceCycles() throws Exception {
+ try (ScenarioFiles files = new ScenarioFiles()) {
+ Path first = files.write("first.yaml", "extends: second.yaml\n");
+ files.write("second.yaml", "extends: ./first.yaml\n");
+ assertThatThrownBy(() -> PulsarProfilingConfig.Config.read(first, Map.of()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("inheritance cycle")
+ .hasMessageContaining("first.yaml")
+ .hasMessageContaining("second.yaml");
+ }
+ }
+
+ @DataProvider
+ public Object[][] invalidInheritance() {
+ return new Object[][] {
+ {"extends: 42\n"},
+ {"extends: ~\n"},
+ {"extends: {}\n"},
+ {"extends: ''\n"},
+ {"extends: [42]\n"},
+ {"extends: [~]\n"},
+ {"extends: missing.yaml\n"},
+ {"- not-a-mapping\n"},
+ {""}
+ };
+ }
+
+ @Test(dataProvider = "invalidInheritance")
+ public void rejectsInvalidInheritance(String yaml) throws Exception {
+ try (ScenarioFiles files = new ScenarioFiles()) {
+ Path child = files.write("scenario.yaml", yaml);
+ assertThatThrownBy(() -> PulsarProfilingConfig.Config.read(child, Map.of()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("config");
+ }
+ }
+
+ @Test
+ public void sharedScenarioOnlyChangesSubscriptionTypeAndOutputDirectory() {
+ Path scenarios = Path.of("../performance/scenarios");
+ var exclusive = PulsarProfilingConfig.Config.read(
+ scenarios.resolve("read-completion-isolation.yaml"), Map.of());
+ var shared = PulsarProfilingConfig.Config.read(scenarios.resolve("read-completion-isolation-shared.yaml"),
+ Map.of());
+ assertThat(shared.cluster()).isEqualTo(exclusive.cluster());
+ assertThat(shared.load()).usingRecursiveComparison().ignoringFields("subscriptionType")
+ .isEqualTo(exclusive.load());
+ assertThat(shared.load().subscriptionType()).isEqualTo(SubscriptionType.Shared);
+ assertThat(shared.output().directory()).isEqualTo("build/pulsar-profiling/read-completion-isolation-shared");
+ }
+
+ private static final class ScenarioFiles implements AutoCloseable {
+ private final Path directory;
+
+ private ScenarioFiles() throws IOException {
+ directory = Files.createTempDirectory("pulsar-profiling-config");
+ }
+
+ private Path write(String name, String yaml) throws IOException {
+ Path path = directory.resolve(name);
+ Files.createDirectories(path.getParent());
+ return Files.writeString(path, yaml);
+ }
+
+ @Override
+ public void close() throws IOException {
+ try (var paths = Files.walk(directory)) {
+ for (Path path : paths.sorted(Comparator.reverseOrder()).toList()) {
+ Files.delete(path);
+ }
+ }
+ }
+ }
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingV4Test.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingV4Test.java
index 37762a866262..a2e6d92c2cfb 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingV4Test.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingV4Test.java
@@ -19,6 +19,8 @@
package org.apache.pulsar.tests.integration.profiling;
import java.util.List;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.common.naming.TopicDomain;
import org.testng.annotations.Test;
@@ -42,6 +44,16 @@ public class PulsarProfilingV4Test extends AbstractPulsarProfilingTest {
return "-v4";
}
+ @Override
+ protected void prepareTopic(String topicName) throws Exception {
+ // Establish the subscription before publishing, even if pulsar-perf takes longer to start.
+ try (PulsarAdmin admin = PulsarAdmin.builder()
+ .serviceHttpUrl(pulsarCluster.getAnyBroker().getHttpServiceUrl()).build()) {
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "sub", MessageId.earliest);
+ }
+ }
+
@Override
protected List<TopicStatsEndpoint> getTopicStatsEndpoints(String topicName) {
String basePath = adminV2Path("persistent", topicName);
diff --git a/tests/performance/README.md b/tests/performance/README.md
index 67289353ad8e..b24955dea202 100644
--- a/tests/performance/README.md
+++ b/tests/performance/README.md
@@ -46,13 +46,13 @@ classic `persistent://` topic. Both variants profile a single broker and write r
output under `tests/integration/build/pulsar-profiling`.
The harness accepts a YAML scenario file through `PULSAR_PROFILING_CONFIG`. Start with
-[`pulsar-profiling.yaml`](pulsar-profiling.yaml); omitted values retain the existing defaults. The
-sections correspond to the main components of a run: `cluster`, `load` and `output`. Individual scalar
+[`pulsar-profiling.yaml`](scenarios/pulsar-profiling.yaml); omitted values retain the existing defaults. The
+sections correspond to the main components of a run: `cluster`, `load`, `profiling` and `output`. Individual scalar
values can still be overridden for a one-off run with the `PULSAR_PROFILING_` prefix and an upper-case
path, for example:
```bash
-PULSAR_PROFILING_CONFIG=tests/performance/pulsar-profiling.yaml \
+PULSAR_PROFILING_CONFIG="$PWD/tests/performance/scenarios/pulsar-profiling.yaml" \
PULSAR_PROFILING_LOAD_NUMBER_OF_MESSAGES=1000000 \
./gradlew :tests:integration:profilingIntegrationTest --tests "*PulsarProfilingV4Test"
```
@@ -60,10 +60,70 @@ PULSAR_PROFILING_LOAD_NUMBER_OF_MESSAGES=1000000 \
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.
+producer test-thread option and with consumer listener-thread expansion. Set `load.producerCount`
+and `load.consumerCount` separately: creating clients does not create producers or consumers.
+`load.subscriptionType` selects the subscription type; `producerIoThreads` and `consumerIoThreads`
+size the shared client IO pools. `maxOutstanding` is per producer, not a global limit.
+Set `load.batchingEnabled: true` to use pulsar-perf's default producer batching; the default is
+`false`, preserving unbatched entry-by-entry measurements.
+
+Client profiling is optional and independent of broker profiling. Set `profiling.producerOptions`
+and/or `profiling.consumerOptions` to async-profiler options, for example
+`event=cpu,interval=10ms,lock=0,alloc=2m,jfrsync=profile`. An empty or null option disables that
+client's profiler. Client recordings are named `client-producer-*.jfr` and `client-consumer-*.jfr`
+in the same output directory. The harness grants native CPU profiling access only to enabled clients.
+The broker continues to use `-Pinttest.asyncprofiler.opts`. To inspect lock contention without
+wall-clock sampling overhead, omit `wall` and use `lock=0` to record all supported lock events.
+Use the same options for baseline and candidate; extra profiling has a measurement cost.
+
+Broker-side variations from the contention investigations need no dedicated environment switches:
+use `cluster.brokerEnvs` for write-buffer watermarks, dispatcher batch size, broker/BookKeeper IO
+thread counts and `PULSAR_GC`; use `cluster.brokerMemory` for heap/direct-memory limits and JVM
+properties such as allocator or transport selection. These maps let YAML scenarios retain the
+complete configuration instead of relying on shell history.
+
+The harness saves `resolved-config.yaml` with inheritance and environment overrides applied in the output directory.
For v4 production, `--num-producers` remains the producer count per topic and is distributed across
the isolated clients; when the counts differ, producers are assigned round-robin as evenly as possible.
+### Inheriting scenario configurations
+
+Use a top-level `extends` to inherit one file or a list of files:
+
+```yaml
+extends: [cluster.yaml, workloads/many-producers.yaml]
+load:
+ subscriptionType: Shared
+cluster:
+ brokerEnvs:
+ preciseDispatcherFlowControl: ~
+output:
+ directory: build/pulsar-profiling/shared
+```
+
+Each path is relative to the file declaring it; absolute paths also work. Parents can themselves
+inherit other files. Starting with the harness defaults, the loader visits each parent recursively
+in the listed order, then applies the current file. Later values win, mappings merge recursively,
+and scalar values and lists replace earlier values. Shared ancestors are applied on each visit;
+inheritance cycles, missing files and invalid `extends` entries are rejected. Environment overrides
+are applied last, and `resolved-config.yaml` contains the resulting values without `extends`.
+
+An explicit YAML `null` or `~` removes an entry, including a harness default. For example, the
+`preciseDispatcherFlowControl` removal above leaves that broker setting to the broker's own default.
+Deleting a mapping removes all its entries; a later mapping starts fresh. Required workload fields
+must still have valid values in the final configuration.
+
+## Reproducible scenarios
+
+- [Read-completion queue isolation](read-completion-isolation.md): 500 producers on separate
+ connections to one persistent topic, with one Exclusive consumer. Includes the
+ [scenario YAML](scenarios/read-completion-isolation.yaml), an inherited
+ [Shared-subscription variant](scenarios/read-completion-isolation-shared.yaml), a
+ [Failover variant](scenarios/read-completion-isolation-failover.yaml), explicit
+ [64/32 KiB](scenarios/read-completion-isolation-64k-32k.yaml) and
+ [256/128 KiB](scenarios/read-completion-isolation-256k-128k.yaml) channel-watermark variations,
+ and baseline/comparison instructions.
+
## Inspecting recordings
Render the CPU, wall-clock, allocation and lock views with:
diff --git a/tests/performance/read-completion-isolation.md b/tests/performance/read-completion-isolation.md
new file mode 100644
index 000000000000..7575799466bf
--- /dev/null
+++ b/tests/performance/read-completion-isolation.md
@@ -0,0 +1,123 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+
+# Read-completion profiling scenario
+
+Use this scenario to measure whether consumer dispatch keeps up when many producers publish to
+one persistent topic. Run the same workload and profiler settings on the base revision and the
+candidate revision. See [README.md](README.md) for the profiling harness and analysis tools.
+
+## Workload
+
+[read-completion-isolation.yaml](scenarios/read-completion-isolation.yaml) runs the v4 `pulsar-perf` commands
+with 500 producers, 500 isolated clients sharing PIP-234 resources, one connection per client,
+and one Exclusive consumer on a non-partitioned `persistent://` topic. It sends 12 million
+unbatched 128-byte messages at unrestricted rate. Each producer permits 40 outstanding sends,
+for a total limit of 20,000. The harness creates the subscription before publishing.
+
+Inherited variations change one aspect of this workload:
+
+- [Shared subscription](scenarios/read-completion-isolation-shared.yaml)
+- [Failover subscription](scenarios/read-completion-isolation-failover.yaml)
+- [64/32 KiB channel high/low watermarks](scenarios/read-completion-isolation-64k-32k.yaml)
+- [256/128 KiB channel high/low watermarks](scenarios/read-completion-isolation-256k-128k.yaml)
+
+Keep inherited YAML files together: parent paths resolve relative to the file declaring them.
+Always use the same variation for the baseline and candidate. A result for one subscription type
+does not establish an improvement for another.
+
+## Run a comparison
+
+Use separate checkouts for the base and candidate revisions. Both must contain the same profiling
+harness and scenario files. If the candidate changes the harness, apply just its harness changes
+to the baseline before measuring. For example, from a candidate checkout based on `origin/master`:
+
+```bash
+git worktree add --detach ../pulsar-read-baseline origin/master
+git diff origin/master...HEAD -- tests/integration tests/performance > /tmp/read-completion-harness.patch
+git -C ../pulsar-read-baseline apply /tmp/read-completion-harness.patch
+```
+
+Run this command from each checkout's root. Replace `baseline` with `candidate` for the candidate
+run and use a fresh output directory for every repetition. Use an absolute YAML path because
+Gradle runs the test from the integration module directory.
+
+```bash
+PULSAR_PROFILING_CONFIG="$PWD/tests/performance/scenarios/read-completion-isolation.yaml" \
+PULSAR_PROFILING_OUTPUT_DIRECTORY="$PWD/tests/integration/build/pulsar-profiling/baseline" \
+./gradlew :tests:integration:profilingIntegrationTest --tests '*PulsarProfilingV4Test' \
+ '-Pinttest.asyncprofiler.opts=event=cpu,interval=10ms,lock=0,alloc=2m,jfrsync=profile'
+```
+
+Build and run the checkouts sequentially because they share Docker image tags. Keep host power
+settings, memory limits and profiler options identical. Avoid other builds or benchmarks during
+measurement; record thermal throttling and available memory. Save the source revision and any local
+diff alongside the output. The harness writes the effective configuration to `resolved-config.yaml`.
+
+To select a variation, change `PULSAR_PROFILING_CONFIG` to its YAML file. For a rate-limited run,
+add `PULSAR_PROFILING_LOAD_PRODUCE_RATE=100000`, which gives 200 messages/s per producer client.
+The message count must be divisible by the number of isolated producer clients.
+
+## Compare results
+
+**Steady dispatch** measures broker delivery while all 500 producers are active. Omit the first
+20 seconds after all producers connect, then use only complete snapshot intervals with 500
+publishers at both endpoints. The window ends before producers start disconnecting. Divide the
+sum of `msgOutCounter` increases by the total interval duration. Calculate **steady ingress**
+the same way from `msgInCounter`. These measure broker activity, not client acknowledgement.
+The cached `msgRateIn` and `msgRateOut` values can lag workload transitions.
+
+**Whole-run throughput** is each `pulsar-perf` client's reported aggregate rate over its own timer.
+Consumer throughput includes ramp-up and backlog drain; the producer timer is separate. A high
+whole-run consumer rate can hide slow dispatch during publishing if the backlog drains quickly
+after producers stop. Compare steady rates and backlog as well as aggregate completion speed.
+
+**Sampled maximum backlog** is the largest `subscriptions.sub.msgBacklog` across all collected
+`stats-v4.*.txt` snapshots. It is a sampled maximum, not an exact peak between snapshots. Report
+the sampling interval alongside it.
+
+Before comparing rates:
+
+- Verify 500 publishers with distinct connection addresses and one consumer of the intended
+ subscription type in the topic stats.
+- Check final client counts and failed ACKs. The consumer must receive 12 million messages;
+ completed-send accounting can let the producer exceed its target by sends already in flight.
+- Reject runs with a timeout, broker restart, OOM or failed persistence, even if clients reconnect.
+- Report percentage changes alongside absolute values and distinguish lower-is-better metrics
+ such as backlog and CPU from higher-is-better throughput.
+
+Analyze the broker recording with Jafar MCP `jfr_diagnose` and `jfr_stackprofile`, saving output
+beside the recording as `<filename>.jfr.analysis.md`. Compare CPU, monitor contention and thread-park
+views within the steady window. Executor queue delay need not appear as monitor contention.
+The command above omits wall-clock sampling to avoid its overhead during lock profiling.
+Compare allocations and actual GC pauses as well; concurrent GC duration is not pause time.
+
+For a focused comparison of completion handoff overhead, run the JMH benchmark separately from the
+integration test:
+
+```bash
+./gradlew :microbench:shadowJar
+java -jar microbench/build/libs/microbench-*-benchmarks.jar \
+ '.*ReadCompletionHandoffBenchmark.*' -prof gc -rf json -rff read-handoff.json
+```
+
+The microbenchmark uses real executors without storage IO or publishing pressure. Its latency and
+allocation results do not predict end-to-end throughput or establish concurrency correctness.
diff --git a/tests/performance/pulsar-profiling.yaml b/tests/performance/scenarios/pulsar-profiling.yaml
similarity index 83%
copy from tests/performance/pulsar-profiling.yaml
copy to tests/performance/scenarios/pulsar-profiling.yaml
index db1316d03a6c..cce6edfff751 100644
--- a/tests/performance/pulsar-profiling.yaml
+++ b/tests/performance/scenarios/pulsar-profiling.yaml
@@ -36,10 +36,22 @@ load:
consumeMemoryLimit: 200M
produceRate: 2147483647
messageSize: 128
+ batchingEnabled: false
maxOutstanding: 20000
statsIntervalSeconds: 10
# Used only by produce-v4/consume-v4. A value of 500 creates up to 500 clients.
+ producerCount: 1
+ consumerCount: 1
+ producerIoThreads: 1
+ consumerIoThreads: 1
+ subscriptionType: Shared
+ receiverQueueSize: 50000
+ timeoutSeconds: 180
isolatedProducers: 0
isolatedConsumers: 0
+profiling:
+ # Empty options leave the client unprofiled. Broker options use -Pinttest.asyncprofiler.opts.
+ producerOptions: ""
+ consumerOptions: ""
output:
directory: build/pulsar-profiling
diff --git a/tests/performance/scenarios/read-completion-isolation-256k-128k.yaml b/tests/performance/scenarios/read-completion-isolation-256k-128k.yaml
new file mode 100644
index 000000000000..674330c8e8b2
--- /dev/null
+++ b/tests/performance/scenarios/read-completion-isolation-256k-128k.yaml
@@ -0,0 +1,25 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+extends: read-completion-isolation.yaml
+cluster:
+ brokerEnvs:
+ pulsarChannelWriteBufferHighWaterMark: "262144"
+ pulsarChannelWriteBufferLowWaterMark: "131072"
+output:
+ directory: build/pulsar-profiling/read-completion-isolation-256k-128k
diff --git a/tests/performance/scenarios/read-completion-isolation-64k-32k.yaml b/tests/performance/scenarios/read-completion-isolation-64k-32k.yaml
new file mode 100644
index 000000000000..a675f60ca463
--- /dev/null
+++ b/tests/performance/scenarios/read-completion-isolation-64k-32k.yaml
@@ -0,0 +1,25 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+extends: read-completion-isolation.yaml
+cluster:
+ brokerEnvs:
+ pulsarChannelWriteBufferHighWaterMark: "65536"
+ pulsarChannelWriteBufferLowWaterMark: "32768"
+output:
+ directory: build/pulsar-profiling/read-completion-isolation-64k-32k
diff --git a/tests/performance/scenarios/read-completion-isolation-failover.yaml b/tests/performance/scenarios/read-completion-isolation-failover.yaml
new file mode 100644
index 000000000000..7a31b5667695
--- /dev/null
+++ b/tests/performance/scenarios/read-completion-isolation-failover.yaml
@@ -0,0 +1,23 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+extends: read-completion-isolation.yaml
+load:
+ subscriptionType: Failover
+output:
+ directory: build/pulsar-profiling/read-completion-isolation-failover
diff --git a/tests/performance/scenarios/read-completion-isolation-shared.yaml b/tests/performance/scenarios/read-completion-isolation-shared.yaml
new file mode 100644
index 000000000000..856fb6e360eb
--- /dev/null
+++ b/tests/performance/scenarios/read-completion-isolation-shared.yaml
@@ -0,0 +1,23 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Same 500-producer workload with one Shared consumer instead of Exclusive.
+extends: read-completion-isolation.yaml
+load:
+ subscriptionType: Shared
+output:
+ directory: build/pulsar-profiling/read-completion-isolation-shared
diff --git a/tests/performance/pulsar-profiling.yaml b/tests/performance/scenarios/read-completion-isolation.yaml
similarity index 56%
rename from tests/performance/pulsar-profiling.yaml
rename to tests/performance/scenarios/read-completion-isolation.yaml
index db1316d03a6c..2090b72e6aea 100644
--- a/tests/performance/pulsar-profiling.yaml
+++ b/tests/performance/scenarios/read-completion-isolation.yaml
@@ -15,31 +15,31 @@
# specific language governing permissions and limitations
# under the License.
-# Pass this file with PULSAR_PROFILING_CONFIG=/path/to/pulsar-profiling.yaml.
-# Values not present here keep the harness defaults.
+# 500 producers / 500 TCP connections / one Exclusive consumer on a persistent topic.
+# Run with PulsarProfilingV4Test; see read-completion-isolation.md for a baseline comparison.
cluster:
numBrokers: 1
numBookies: 3
numProxies: 0
- brokerMemory: "-Xms2g -Xmx2g -XX:+UseTransparentHugePages -XX:+AlwaysPreTouch"
- bookkeeperMemory: "-Xmx1g -XX:+UseTransparentHugePages -XX:+AlwaysPreTouch"
- brokerEnvs:
- managedLedgerDefaultEnsembleSize: "1"
- managedLedgerDefaultWriteQuorum: "1"
- managedLedgerDefaultAckQuorum: "1"
- bookkeeperEnvs:
- dbStorage_writeCacheMaxSizeMb: "64"
- dbStorage_readAheadCacheMaxSizeMb: "96"
+ brokerMemory: "-Xms2g -Xmx2g -XX:MaxDirectMemorySize=4g"
+ bookkeeperMemory: "-Xms512m -Xmx512m -XX:MaxDirectMemorySize=512m"
load:
- numberOfMessages: 20000000
+ numberOfMessages: 12000000
+ producerCount: 500
+ isolatedProducers: 500
+ producerIoThreads: 8
+ consumerCount: 1
+ isolatedConsumers: 0
+ consumerIoThreads: 1
+ subscriptionType: Exclusive
+ receiverQueueSize: 50000
+ # This is a per-producer queue limit: 40 * 500 = 20000 outstanding sends.
+ maxOutstanding: 40
produceMemoryLimit: 200M
consumeMemoryLimit: 200M
produceRate: 2147483647
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
+ statsIntervalSeconds: 5
+ timeoutSeconds: 480
output:
- directory: build/pulsar-profiling
+ directory: build/pulsar-profiling/read-completion-isolation