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 09:21:40 UTC
(pulsar) 01/01: [improve][test] Add YAML scenarios to profiling harness
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-perfopt-profiling-yaml-config
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6b25dfcbd06ffba587f6d846759da03ff89dd98e
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Sep 16 12:09:57 2026 +0300
[improve][test] Add YAML scenarios to profiling harness
---
CONTRIBUTING.md | 53 +-----
microbench/README.md | 4 +-
tests/README.md | 2 +-
.../profiling/AbstractPulsarProfilingTest.java | 73 +++-----
.../profiling/PulsarProfilingConfig.java | 183 +++++++++++++++++++++
.../profiling/PulsarProfilingConfigTest.java | 45 +++++
tests/performance/README.md | 99 +++++++++++
tests/performance/pulsar-profiling.yaml | 42 +++++
8 files changed, 403 insertions(+), 98 deletions(-)
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index b1a24db8d922..eb09e79df9ae 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -283,56 +283,11 @@ disallows privileged containers (the run continues either way, with less accurat
`-Pdocker.wolfi` builds the base image from Wolfi, which is what makes the `GLIBC_TUNABLES` the test
sets take effect.
-#### Analyzing a JFR file
+#### Performance recording analysis
-A recording collects several events at once, and each of them is worth looking at from more than one
-angle, so **`jfrFlamegraphs`** renders the lot in one go:
-
-```bash
-./gradlew jfrFlamegraphs # every .jfr in build/test-profiles
-./gradlew jfrFlamegraphs -Pjfr=tests/integration/build # the integration-test recordings
-./gradlew jfrFlamegraphs -Pjfr=<some-file>.jfr # just one recording
-```
-
-Each recording gets a directory beside it named after the file without its extension plus a
-`-flamegraphs` suffix — `profile.jfr` produces `profile-flamegraphs/` — holding one flame graph per
-view: `cpu`, `wall`, `alloc` and `lock`, each rendered merged (`cpu.html`), split per thread
-(`cpu_threads.html`) and grouped into async-profiler's categories (`cpu_classify.html`). A view whose
-event the recording does not contain is skipped rather than failing the run. Pass
-`-Pjfr.types=cpu,nativemem` to render a different set of events, and `-Pjfrconv=<path>` when
-async-profiler's `jfrconv` is neither next to the library `LIBASYNCPROFILER_PATH` points at nor in
-the JDK that runs Gradle (Amazon Corretto ships it).
-
-For anything the flame graphs don't answer, open the `.jfr` itself in
-[Eclipse Mission Control](https://adoptium.net/jmc) or IntelliJ IDEA, or run `jfrconv` by hand.
-
-> Do not judge a recording by `jfr summary` (the JDK's own tool). The default options record through
-> `jfrsync`, and `jfr summary` reports only a handful of `jdk.ExecutionSample` events for a recording
-> that `jfrconv` reads hundreds of profiler samples from — it looks empty when it is not.
-
-#### Agent-assisted analysis with the Jafar MCP server
-
-The [Jafar MCP server](https://github.com/btraceio/jafar/blob/main/jfr-mcp/README.md) lets an AI
-coding agent read a JFR recording directly, which turns a flame graph into something you can ask
-questions about. Register it once (it needs [JBang](https://www.jbang.dev/) and JDK 25+):
-
-```bash
-claude mcp add jafar -- jbang jfr-mcp@btraceio --stdio
-```
-
-It exposes `jfr_diagnose` (automated diagnosis of a recording), `jfr_stackprofile` (structured stack
-profiling with a time-series and per-thread breakdown), `jfr_hotmethods`, `jfr_flamegraph`,
-`jfr_callgraph`, `jfr_exceptions`, `jfr_tsa` (thread-state analysis), `jfr_use` (USE method) and
-`jfr_query` for [JfrPath](https://github.com/btraceio/jafar) queries. A prompt that works well as a
-starting point:
-
-> use Jafar MCP's jfr_diagnose and jfr_stackprofile to analyze @filename.jfr. Besides showing the
-> report on the console, write the analysis in a markdown file with the jfr file as prefix and the
-> suffix as ".analysis.md"
-
-Treat the result as a lead to verify, not a conclusion: the
-[agent guardrails](AGENTS.md) apply here as much as anywhere, and a performance claim still needs a
-benchmark or a second profile behind it.
+See [`tests/performance/README.md`](tests/performance/README.md) for JFR rendering, async-profiler
+recording analysis, Jafar MCP usage and MAT MCP based memory-leak investigation. That document also
+explains how to keep the raw recordings and analysis next to the workload documentation.
### Integration tests
diff --git a/microbench/README.md b/microbench/README.md
index 1704cdd9a9cb..0dbd923dee6f 100644
--- a/microbench/README.md
+++ b/microbench/README.md
@@ -133,10 +133,10 @@ Each recording gets a directory beside it named after the file without its exten
`-flamegraphs` suffix, holding `cpu`, `wall`, `alloc` and `lock`, each rendered merged
(`cpu.html`), split per thread (`cpu_threads.html`) and grouped into async-profiler's categories
(`cpu_classify.html`). A view whose event the recording does not contain is skipped. See
-[Analyzing a JFR file](../CONTRIBUTING.md#analyzing-a-jfr-file) for the options it takes.
+[Performance testing](../tests/performance/README.md) for the analysis workflow.
The `.jfr` can also be opened in [Eclipse Mission Control](https://adoptium.net/jmc) or IntelliJ
IDEA, or handed to an AI agent through the
[Jafar MCP server](https://github.com/btraceio/jafar/blob/main/jfr-mcp/README.md), which lets the
agent query the recording directly with tools such as `jfr_diagnose` and `jfr_stackprofile` — see
-[Agent-assisted analysis](../CONTRIBUTING.md#agent-assisted-analysis-with-the-jafar-mcp-server).
\ No newline at end of file
+[Jafar MCP analysis](../tests/performance/README.md#jafar-mcp-analysis).
diff --git a/tests/README.md b/tests/README.md
index 6d46413b5902..de76643700f1 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -34,4 +34,4 @@ The directories are as follows:
- docker-images/ : Docker images for integration testing.
- integration/ : The integration tests and utilities themselves.
-
+- performance/ : Scenario files and analysis guidance for repeatable profiling runs.
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 7866fec2fe55..0f5d9f48eb61 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
@@ -58,7 +58,7 @@ import org.testcontainers.containers.GenericContainer;
* That single task builds the test image with async-profiler in it, relaxes the kernel perf_event
* limits that the cpu sampling engine needs, and runs {@link PulsarProfilingTest} against the
* result. Add --tests "*PulsarProfilingV4Test" to profile the v4 variant instead. See
- * CONTRIBUTING.md for the properties that tune it.
+ * {@code tests/performance/README.md} for scenario YAML files and environment overrides.
* On a Linux host the perf_event limits can also be set persistently with sysctl, in which case
* -Pinttest.asyncprofiler.skipPerfEventTuning skips the container that sets them:
* kernel.perf_event_paranoid=1
@@ -80,13 +80,9 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
// echo madvise | sudo tee /sys/kernel/mm/transparent_hugepage/defrag
// More info about -XX:+UseTransparentHugePages at
// https://shipilev.net/jvm/anatomy-quarks/2-transparent-huge-pages/
- private static final String DEFAULT_PULSAR_MEM = "-Xmx1g -XX:+UseTransparentHugePages -XX:+AlwaysPreTouch";
private static final String PERF_PULSAR_MEM =
"-XX:+UseTransparentHugePages -XX:+AlwaysPreTouch -XX:+HeapDumpOnOutOfMemoryError "
+ "-XX:HeapDumpPath=/testoutput/%HEAP_DUMP_NAME%";
- private static final String PRODUCE_MEM_LIMIT = "200M";
- private static final String CONSUME_MEM_LIMIT = "200M";
- private static final String BROKER_PULSAR_MEM = "-Xms2g -Xmx2g -XX:+UseTransparentHugePages -XX:+AlwaysPreTouch";
/**
* The topic domain to drive the load against: {@link TopicDomain#topic} for a v5 scalable topic,
@@ -130,24 +126,26 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
return "/admin/v2/" + resource + "/" + topic.getNamespace() + "/" + topic.getLocalName();
}
- // A container that runs pulsar-perf, arguments are currently hard-coded since this is an example
+ // A container that runs pulsar-perf for the configured profiling scenario.
static class PulsarPerfContainer extends GenericContainer<PulsarPerfContainer> {
private final String brokerHostname;
private final String commandSuffix;
+ private final PulsarProfilingConfig.Load load;
// Sized to finish well inside the wait in runPulsarPerfBenchmark. The containers sustain
// roughly 290k msg/s, so this is a bit over a minute of load - long enough for a profile,
// short enough that a run which does not finish is a real stall rather than the normal end.
- private final long numberOfMessages = 20_000_000;
public PulsarPerfContainer(File testOutputDir,
String clusterName,
String brokerHostname,
String hostname,
String memArgs,
- String commandSuffix) {
+ String commandSuffix,
+ PulsarProfilingConfig.Load load) {
super(PulsarContainer.DEFAULT_IMAGE_NAME);
this.brokerHostname = brokerHostname;
this.commandSuffix = commandSuffix;
+ this.load = load;
withCreateContainerCmdModifier(createContainerCmd -> {
createContainerCmd.withHostName(hostname);
createContainerCmd.withName(clusterName + "-" + hostname);
@@ -166,7 +164,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
+ "-u pulsar://" + brokerHostname + ":6650 "
+ "-st Shared "
+ "-q 50000 "
- + "-m " + numberOfMessages + " -ml " + CONSUME_MEM_LIMIT + " "
+ + "-m " + load.numberOfMessages() + " -ml " + load.consumeMemoryLimit() + " "
+ "--histogram-file=/testoutput/consume" + commandSuffix
+ ".histogram.$(date +%s).hdr "
+ "2>&1 | tee /testoutput/consume" + commandSuffix + ".$(date +%s).txt");
@@ -178,12 +176,12 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
+ "/pulsar/bin/pulsar-perf produce" + commandSuffix + " " + topicName + " "
+ "-u pulsar://" + brokerHostname + ":6650 "
+ "-au http://" + brokerHostname + ":8080 "
- + "-r " + Integer.MAX_VALUE + " "
- + "-s 128 -db "
+ + "-r " + load.produceRate() + " "
+ + "-s " + load.messageSize() + " -db "
// maxOutstanding only applies to the v4 client; the v5 client accepts
// the flag for back-compat but ignores it
- + "-o 20000 "
- + "-m " + numberOfMessages + " -ml " + PRODUCE_MEM_LIMIT + " "
+ + "-o " + load.maxOutstanding() + " "
+ + "-m " + load.numberOfMessages() + " -ml " + load.produceMemoryLimit() + " "
+ "--histogram-file=/testoutput/produce" + commandSuffix
+ ".histogram.$(date +%s).hdr "
+ "2>&1 | tee /testoutput/produce" + commandSuffix + ".$(date +%s).txt");
@@ -207,7 +205,8 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
.append(commandSuffix).append(".$(date +%s).txt; ");
}
script.append("curl -s ").append(brokerUrl).append("/metrics/ > /testoutput/metrics")
- .append(commandSuffix).append(".$(date +%s).txt; sleep 10; done");
+ .append(commandSuffix).append(".$(date +%s).txt; sleep ")
+ .append(load.statsIntervalSeconds()).append("; done");
return DockerUtils.runCommandAsyncWithLogging(getDockerClient(), getContainerId(),
"bash", "-c", script.toString());
}
@@ -240,6 +239,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
private PulsarPerfContainer perfProduce;
private PulsarPerfContainer printStats;
private File testOutputDir;
+ private final PulsarProfilingConfig.Config profilingConfig = PulsarProfilingConfig.Config.read();
@Override
public void setupCluster() throws Exception {
@@ -249,7 +249,7 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
}
private void createTestOutputDir() {
- testOutputDir = new File("build/pulsar-profiling");
+ testOutputDir = new File(profilingConfig.output().directory());
if (!testOutputDir.exists()) {
if (!testOutputDir.mkdirs()) {
throw new IllegalArgumentException("Test output directory + '" + testOutputDir.getAbsolutePath()
@@ -315,58 +315,39 @@ public abstract class AbstractPulsarProfilingTest extends PulsarTestSuite {
specBuilder.profileDirectory(testOutputDir.getAbsolutePath());
// Only run one broker so that all load goes to a single broker
- specBuilder.numBrokers(1);
+ PulsarProfilingConfig.Cluster cluster = profilingConfig.cluster();
+ specBuilder.numBrokers(cluster.numBrokers());
// Have 3 bookies to reduce bottleneck on bookie
- specBuilder.numBookies(3);
+ specBuilder.numBookies(cluster.numBookies());
// no need for proxy
- specBuilder.numProxies(0);
+ specBuilder.numProxies(cluster.numProxies());
// Increase memory for brokers and configure more aggressive rollover
- Map<String, String> brokerEnvs = new HashMap<>();
- brokerEnvs.put("PULSAR_MEM", BROKER_PULSAR_MEM);
- brokerEnvs.put("managedLedgerMinLedgerRolloverTimeMinutes", "1");
- brokerEnvs.put("managedLedgerMaxLedgerRolloverTimeMinutes", "5");
- brokerEnvs.put("managedLedgerMaxSizePerLedgerMbytes", "512");
- brokerEnvs.put("managedLedgerDefaultEnsembleSize", "1");
- brokerEnvs.put("managedLedgerDefaultWriteQuorum", "1");
- brokerEnvs.put("managedLedgerDefaultAckQuorum", "1");
+ Map<String, String> brokerEnvs = new HashMap<>(cluster.brokerEnvs());
+ brokerEnvs.put("PULSAR_MEM", cluster.brokerMemory());
//brokerEnvs.put("maxPendingPublishRequestsPerConnection", "1000");
- brokerEnvs.put("dispatcherRetryBackoffInitialTimeInMs", "0");
- brokerEnvs.put("dispatcherRetryBackoffMaxTimeInMs", "0");
- brokerEnvs.put("preciseDispatcherFlowControl", "true");
//brokerEnvs.put("PULSAR_PREFIX_subscriptionKeySharedUseClassicPersistentImplementation", "true");
//brokerEnvs.put("PULSAR_PREFIX_subscriptionSharedUseClassicPersistentImplementation", "true");
- brokerEnvs.put("dispatcherMaxReadBatchSize", "1000");
//brokerEnvs.put("dispatcherMaxReadSizeBytes", "10000000");
//brokerEnvs.put("dispatcherDispatchMessagesInSubscriptionThread", "false");
//brokerEnvs.put("dispatcherMaxRoundRobinBatchSize", "1000");
specBuilder.brokerEnvs(brokerEnvs);
// Increase memory for bookkeepers and make compaction run more often
- Map<String, String> bkEnv = new HashMap<>();
- bkEnv.put("PULSAR_MEM", DEFAULT_PULSAR_MEM);
- bkEnv.put("dbStorage_writeCacheMaxSizeMb", "64");
- bkEnv.put("dbStorage_readAheadCacheMaxSizeMb", "96");
- bkEnv.put("journalMaxSizeMB", "256");
- bkEnv.put("journalSyncData", "false");
- bkEnv.put("majorCompactionInterval", "300");
- bkEnv.put("minorCompactionInterval", "30");
- bkEnv.put("compactionRateByEntries", "20000");
- bkEnv.put("gcWaitTime", "30000");
- bkEnv.put("isForceGCAllowWhenNoSpace", "true");
- bkEnv.put("diskUsageLwmThreshold", "0.75");
- bkEnv.put("diskCheckInterval", "60");
+ Map<String, String> bkEnv = new HashMap<>(cluster.bookkeeperEnvs());
+ bkEnv.put("PULSAR_MEM", cluster.bookkeeperMemory());
specBuilder.bookkeeperEnvs(bkEnv);
// Create pulsar-perf containers
String brokerHostname = clusterName + "-pulsar-broker-0";
String commandSuffix = getPerfCommandSuffix();
+ PulsarProfilingConfig.Load load = profilingConfig.load();
perfProduce = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "perf-produce", "-Xmx2g",
- commandSuffix);
+ commandSuffix, load);
perfConsume = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "perf-consume", "-Xmx1g",
- commandSuffix);
+ commandSuffix, load);
printStats = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "print-stats", "-Xmx1g",
- commandSuffix);
+ commandSuffix, load);
specBuilder.externalServices(Map.of(
"pulsar-produce", perfProduce,
"pulsar-consume", perfConsume,
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
new file mode 100644
index 000000000000..230a3d7d5664
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfig.java
@@ -0,0 +1,183 @@
+/*
+ * 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.pulsar.tests.integration.profiling;
+
+import com.fasterxml.jackson.databind.JsonNode;
+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.Map;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+
+/** Configuration for the profiling scenario harness. */
+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) {
+ static Config read() {
+ String configFile = System.getenv(CONFIG_ENV);
+ return read(configFile == null || configFile.isBlank() ? null : Path.of(configFile),
+ System.getenv());
+ }
+
+ static Config read(Path configFile, Map<String, String> environment) {
+ 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);
+ }
+ }
+ applyEnvironmentOverrides(root, environment);
+ try {
+ return mapper.treeToValue(root, Config.class);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Cannot parse profiling configuration", e);
+ }
+ }
+
+ static Config defaults() {
+ return new Config(
+ new Cluster(1, 3, 0,
+ "-Xms2g -Xmx2g -XX:+UseTransparentHugePages -XX:+AlwaysPreTouch",
+ "-Xmx1g -XX:+UseTransparentHugePages -XX:+AlwaysPreTouch",
+ Map.ofEntries(
+ Map.entry("managedLedgerMinLedgerRolloverTimeMinutes", "1"),
+ Map.entry("managedLedgerMaxLedgerRolloverTimeMinutes", "5"),
+ Map.entry("managedLedgerMaxSizePerLedgerMbytes", "512"),
+ Map.entry("managedLedgerDefaultEnsembleSize", "1"),
+ Map.entry("managedLedgerDefaultWriteQuorum", "1"),
+ Map.entry("managedLedgerDefaultAckQuorum", "1"),
+ Map.entry("dispatcherRetryBackoffInitialTimeInMs", "0"),
+ Map.entry("dispatcherRetryBackoffMaxTimeInMs", "0"),
+ Map.entry("preciseDispatcherFlowControl", "true"),
+ Map.entry("dispatcherMaxReadBatchSize", "1000")),
+ Map.ofEntries(
+ Map.entry("dbStorage_writeCacheMaxSizeMb", "64"),
+ Map.entry("dbStorage_readAheadCacheMaxSizeMb", "96"),
+ Map.entry("journalMaxSizeMB", "256"),
+ Map.entry("journalSyncData", "false"),
+ Map.entry("majorCompactionInterval", "300"),
+ Map.entry("minorCompactionInterval", "30"),
+ Map.entry("compactionRateByEntries", "20000"),
+ Map.entry("gcWaitTime", "30000"),
+ 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 Output("build/pulsar-profiling"));
+ }
+ }
+
+ record Cluster(int numBrokers, int numBookies, int numProxies, String brokerMemory,
+ String bookkeeperMemory, Map<String, String> brokerEnvs,
+ Map<String, String> bookkeeperEnvs) {
+ }
+
+ record Load(long numberOfMessages, String produceMemoryLimit, String consumeMemoryLimit,
+ int produceRate, int messageSize, int maxOutstanding, int statsIntervalSeconds) {
+ }
+
+ record Output(String directory) {
+ }
+
+ private static void merge(ObjectNode target, JsonNode source) {
+ if (source == null || !source.isObject()) {
+ return;
+ }
+ source.properties().forEach(entry -> {
+ JsonNode current = target.get(entry.getKey());
+ if (current != null && current.isObject() && entry.getValue().isObject()) {
+ merge((ObjectNode) current, entry.getValue());
+ } else {
+ target.set(entry.getKey(), entry.getValue());
+ }
+ });
+ }
+
+ private static void applyEnvironmentOverrides(ObjectNode root, Map<String, String> environment) {
+ ObjectMapper mapper = ObjectMapperFactory.getYamlMapper().getObjectMapper();
+ environment.forEach((name, value) -> {
+ if (!name.startsWith(ENV_PREFIX) || name.equals(CONFIG_ENV)) {
+ return;
+ }
+ String[] path = name.substring(ENV_PREFIX.length()).toLowerCase().split("_");
+ ObjectNode node = root;
+ int pathIndex = 0;
+ while (pathIndex < path.length) {
+ String field = findField(node, path, pathIndex);
+ if (field == null) {
+ return;
+ }
+ int consumed = field.split("(?=[A-Z])").length;
+ JsonNode existing = node.get(field);
+ if (pathIndex + consumed == path.length) {
+ node.set(field, parseValue(mapper, value, existing));
+ return;
+ }
+ if (!(existing instanceof ObjectNode)) {
+ return;
+ }
+ node = (ObjectNode) existing;
+ pathIndex += consumed;
+ }
+ });
+ }
+
+ private static String findField(ObjectNode node, String[] path, int start) {
+ StringBuilder candidate = new StringBuilder();
+ String result = null;
+ int resultLength = 0;
+ int tokenCount = 0;
+ for (int i = start; i < path.length; i++) {
+ candidate.append(path[i]);
+ tokenCount++;
+ String candidateName = candidate.toString();
+ var fields = node.fieldNames();
+ while (fields.hasNext()) {
+ String field = fields.next();
+ if (field.replace("_", "").equalsIgnoreCase(candidateName)) {
+ result = field;
+ resultLength = tokenCount;
+ }
+ }
+ }
+ return result;
+ }
+
+ private static JsonNode parseValue(ObjectMapper mapper, String value, JsonNode existing) {
+ if (existing.isBoolean()) {
+ return mapper.getNodeFactory().booleanNode(Boolean.parseBoolean(value));
+ }
+ if (existing.isIntegralNumber()) {
+ return mapper.getNodeFactory().numberNode(Long.parseLong(value));
+ }
+ if (existing.isFloatingPointNumber()) {
+ return mapper.getNodeFactory().numberNode(Double.parseDouble(value));
+ }
+ return mapper.getNodeFactory().textNode(value);
+ }
+
+ private PulsarProfilingConfig() {
+ }
+}
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
new file mode 100644
index 000000000000..472d8f6a4894
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingConfigTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.pulsar.tests.integration.profiling;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+public class PulsarProfilingConfigTest {
+ @Test
+ public void loadsYamlAndEnvironmentOverrides() throws Exception {
+ Path file = Files.createTempFile("pulsar-profiling", ".yaml");
+ try {
+ Files.writeString(file, "load:\n messageSize: 256\noutput:\n directory: custom-output\n");
+ PulsarProfilingConfig.Config config = PulsarProfilingConfig.Config.read(file, Map.of(
+ "PULSAR_PROFILING_LOAD_NUMBER_OF_MESSAGES", "1234",
+ "PULSAR_PROFILING_CLUSTER_NUM_BOOKIES", "5"));
+
+ assertThat(config.load().messageSize()).isEqualTo(256);
+ assertThat(config.load().numberOfMessages()).isEqualTo(1234);
+ assertThat(config.cluster().numBookies()).isEqualTo(5);
+ assertThat(config.output().directory()).isEqualTo("custom-output");
+ } finally {
+ Files.deleteIfExists(file);
+ }
+ }
+}
diff --git a/tests/performance/README.md b/tests/performance/README.md
new file mode 100644
index 000000000000..6510512f5955
--- /dev/null
+++ b/tests/performance/README.md
@@ -0,0 +1,99 @@
+<!--
+
+ 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.
+
+-->
+
+# Performance testing
+
+This directory documents repeatable performance experiments and their analysis. The container based
+profiling harness lives under [`tests/integration`](../integration); its recordings normally land in
+`tests/integration/build/pulsar-profiling`. Keep scenario files, commands, results and interpretation
+here so that a later run can reproduce the same workload.
+
+For micro-level questions about one class or method, use the JMH benchmarks in
+[`microbench`](../../microbench). JMH is the benchmark harness; this directory is for documenting the
+end-to-end profiling scenario, profile collection, analysis and conclusions. A useful experiment keeps
+the workload definition, the revision under test, the profiler options, the raw recording and the
+resulting analysis together.
+
+## Profiling an integration-test cluster
+
+Run the built-in scenarios with:
+
+```bash
+./gradlew :tests:integration:profilingIntegrationTest
+./gradlew :tests:integration:profilingIntegrationTest --tests "*PulsarProfilingV4Test"
+```
+
+The first command profiles the v5 scalable-topic scenario. The second uses the v4 client against a
+classic `persistent://` topic. Both variants profile a single broker and write recordings and command
+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
+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_LOAD_NUMBER_OF_MESSAGES=1000000 \
+./gradlew :tests:integration:profilingIntegrationTest --tests "*PulsarProfilingV4Test"
+```
+
+## Inspecting recordings
+
+Render the CPU, wall-clock, allocation and lock views with:
+
+```bash
+./gradlew jfrFlamegraphs -Pjfr=tests/integration/build/pulsar-profiling
+```
+
+The `.jfr` files can also be opened in [Eclipse Mission Control](https://adoptium.net/jmc) or IntelliJ
+IDEA. Do not use `jfr summary` as a measure of profile completeness: recordings made with
+`jfrsync=profile` contain profiler samples that the JDK summary does not show.
+
+### Jafar MCP analysis
+
+The [Jafar MCP server](https://github.com/btraceio/jafar/blob/main/jfr-mcp/README.md) lets an AI coding
+agent query a recording. Register it once with [JBang](https://www.jbang.dev/) and JDK 25+:
+
+```bash
+claude mcp add jafar -- jbang jfr-mcp@btraceio --stdio
+```
+
+Use `jfr_diagnose` and `jfr_stackprofile` first, then query further with the other Jafar tools when
+needed. Save the result beside the recording as `<recording>.analysis.md`, in addition to showing the
+report in the console. A useful starting prompt is:
+
+> use Jafar MCP's jfr_diagnose and jfr_stackprofile to analyze @filename.jfr. Besides showing the
+> report on the console, write the analysis in a markdown file with the jfr file as prefix and the
+> suffix as ".analysis.md"
+
+Treat automated analysis as a lead. Confirm a performance claim with a controlled comparison, a JMH
+benchmark where appropriate, or a second profile.
+
+### Heap dumps and memory leaks with MAT MCP
+
+For an `OutOfMemoryError` or suspected retention problem, analyze the resulting `.hprof` with a
+headless Eclipse Memory Analyzer (MAT) MCP server such as
+[`mcp-mat`](https://github.com/codelipenghui/mcp-mat). Use the leak suspects report and dominator tree first,
+then query paths to GC roots or OQL for the retained objects. Keep the heap dump and the MCP result
+outside the source tree when they contain sensitive workload data; record the commands, heap limits and
+the resulting conclusions in the experiment notes.
diff --git a/tests/performance/pulsar-profiling.yaml b/tests/performance/pulsar-profiling.yaml
new file mode 100644
index 000000000000..0cd8b528e3cd
--- /dev/null
+++ b/tests/performance/pulsar-profiling.yaml
@@ -0,0 +1,42 @@
+# 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.
+
+# Pass this file with PULSAR_PROFILING_CONFIG=/path/to/pulsar-profiling.yaml.
+# Values not present here keep the harness defaults.
+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"
+load:
+ numberOfMessages: 20000000
+ produceMemoryLimit: 200M
+ consumeMemoryLimit: 200M
+ produceRate: 2147483647
+ messageSize: 128
+ maxOutstanding: 20000
+ statsIntervalSeconds: 10
+output:
+ directory: build/pulsar-profiling