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/04 21:03:11 UTC
(pulsar) branch master updated: [improve][build] Profile Pulsar tests with async-profiler from Gradle (#26460)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 23fa956aacc5 [improve][build] Profile Pulsar tests with async-profiler from Gradle (#26460)
23fa956aacc5 is described below
commit 23fa956aacc5d7195d2229dba69ec71605fb28e1
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Sep 5 00:03:01 2026 +0300
[improve][build] Profile Pulsar tests with async-profiler from Gradle (#26460)
---
CODING.md | 2 +
CONTRIBUTING.md | 178 +++++++++++++++++++++
.../src/main/kotlin/JfrFlamegraphsTask.kt | 132 +++++++++++++++
.../main/kotlin/pulsar.java-conventions.gradle.kts | 107 +++++++++++++
build.gradle.kts | 47 ++++++
.../org/apache/pulsar/tests/ManualTestUtil.java | 18 ++-
buildtools/src/main/resources/log4j2.xml | 2 +-
gradle/libs.versions.toml | 1 +
microbench/README.md | 44 ++++-
.../cache/BrokerEntryCacheRollingRestartTest.java | 4 +-
tests/docker-images/java-test-image/Dockerfile | 7 +-
.../docker-images/java-test-image/build.gradle.kts | 50 ++++--
.../docker-images/latest-version-image/Dockerfile | 7 +-
.../latest-version-image/build.gradle.kts | 4 +
tests/integration/build.gradle.kts | 161 +++++++++++++++++--
.../integration/containers/PulsarContainer.java | 53 ++++--
.../integration/profiling/PulsarProfilingTest.java | 85 +++++-----
.../integration/topologies/PulsarCluster.java | 22 ++-
.../integration/topologies/PulsarClusterSpec.java | 40 ++++-
19 files changed, 849 insertions(+), 115 deletions(-)
diff --git a/CODING.md b/CODING.md
index 892b5a051687..75dace9439a4 100644
--- a/CODING.md
+++ b/CODING.md
@@ -236,6 +236,8 @@ defaulting to the safe/old behaviour.
- **Back optimizations with evidence** — a JMH benchmark (see *Testing conventions*) or a profile, not
intuition — measured on JIT-warmed code (see *Reproducing concurrency / memory-visibility bugs*).
+ `-PtestAsyncProfiler` profiles a test run with async-profiler (see
+ [`CONTRIBUTING.md`](CONTRIBUTING.md#profiling-tests-with-async-profiler)).
- **On hot paths** (dispatch, IO, per-message): avoid `String.format` (build strings directly),
`Enum.values()` (match explicitly), and unnecessary allocation/locking; prefer lock-free or
single-writer designs.
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 88b0cfa84470..fa43a62b682b 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -109,6 +109,184 @@ Failed tests are retried once by default (`testRetryCount=1`; `0` when running i
running tests locally, prefer **`-PtestRetryCount=0`** to catch failures (including flakiness) early
instead of having retries mask them.
+### Micro benchmarks (JMH)
+
+For a **micro**-level question — what a single method, data structure or codec costs — write a
+[JMH](https://openjdk.org/projects/code-tools/jmh/) benchmark under `microbench/`. That is the
+preferred tool for micro benchmarks: JMH handles warm-up, dead-code elimination and measurement, none
+of which hand-written timing code gets right. [`CODING.md`](CODING.md#performance) asks for
+optimizations to be backed by evidence, and for a small self-contained change a benchmark is the
+strongest kind.
+
+```bash
+./gradlew :microbench:shadowJar # build the runnable jar
+java -jar microbench/build/libs/microbench-*-benchmarks.jar -l # list the benchmarks
+java -jar microbench/build/libs/microbench-*-benchmarks.jar ".*<Name>.*" # run the ones that match
+```
+
+A benchmark can be profiled directly through JMH's async-profiler integration, which writes forward
+and reverse flame graphs per benchmark under `dir=`:
+
+```bash
+export LIBASYNCPROFILER_PATH=$(ls $JAVA_HOME/lib/libasyncProfiler.*) # Corretto ships one
+java -jar microbench/build/libs/microbench-*-benchmarks.jar \
+ -prof async:libPath=$LIBASYNCPROFILER_PATH\;output=flamegraph\;dir=profile-results ".*<Name>.*"
+```
+
+See [`microbench/README.md`](microbench/README.md) for the rest: the async-profiler setup when the
+JDK does not ship one, recording a benchmark to JFR and rendering it, JSON result files for
+[JMH Visualizer](https://jmh.morethan.io/), and the `rawCommand` escape hatch for async-profiler
+options the JMH plugin does not expose.
+
+### Profiling tests with async-profiler
+
+Profiling a *test run* answers the question a micro benchmark cannot: not what one method costs, but
+where a broker actually spends its time end to end.
+
+> **When the numbers matter, profile and benchmark on Linux x86_64.** That is Pulsar's most common
+> deployment target, and results from elsewhere do not carry over. Two differences bite in
+> particular: async-profiler supports only some of its sampling engines on macOS, so a profile taken
+> there is less reliable; and `System.nanoTime()` is far more expensive on macOS than on Linux, which
+> skews the results in some cases — code that times itself, and JMH's own measurement loop, both pay
+> that cost. Working on macOS or arm64 is fine for finding your way around the code — just treat what
+> it tells you as provisional until it is confirmed on Linux x86_64.
+
+#### Profiling a module's tests
+
+Pass **`-PtestAsyncProfiler`** to run a test task under
+[async-profiler](https://github.com/async-profiler/async-profiler). The build finds the profiler
+library in `LIBASYNCPROFILER_PATH`, and falls back to the copy that Amazon Corretto ships inside the
+JDK that runs the tests, so on Corretto no setup is needed at all:
+
+```bash
+# only needed when the JDK does not ship async-profiler
+export LIBASYNCPROFILER_PATH=$(ls $JAVA_HOME/lib/libasyncProfiler.*)
+./gradlew :pulsar-broker:test -PtestAsyncProfiler --tests "<SomeTest>"
+```
+
+Enabling it runs the tests in a single JVM with retries off and a pre-touched fixed-size heap, and
+sends log4j output to a file, so that one run produces one profile that isn't distorted by console
+logging or by heap resizing. The task always re-runs and is never cached. **Manual tests are enabled
+automatically** whenever profiling is on — the long-running, load-generating tests that are otherwise
+skipped are usually the ones worth profiling — so `ENABLE_MANUAL_TEST` does not have to be exported. Recordings are written to
+`build/test-profiles/` in the repository root, named after the test task and stamped with the start
+time and the pid — for example `test_profile_pulsar-broker-test_20260904-114040_23420.jfr`, next to
+`test_profile_pulsar-broker-test.log`. See [Analyzing a JFR file](#analyzing-a-jfr-file) below.
+
+The defaults can be tuned with `-Ptest.asyncprofiler.event=<event>` (the CPU sampling engine:
+`cpu` on Linux, `itimer` elsewhere — see
+[CPU sampling engines](https://github.com/async-profiler/async-profiler/blob/master/docs/CpuSamplingEngines.md)),
+`-Ptest.asyncprofiler.opts=<agent options>` (default `event=<event>,all,alloc=2m,jfrsync=profile`),
+`-Ptest.asyncprofiler.outputformat=jfr|html|collapsed`, `-Ptest.asyncprofiler.dir=<dir>` (use it to
+keep the profiles of an A/B comparison apart) and `-Ptest.asyncprofiler.libpath=<path>`.
+
+`outputformat` only sets the file extension; the format itself comes from the agent options, and the
+default `jfrsync=profile` forces a JFR recording. To get something other than JFR, override both — to
+profile exception creation, for example (Linux only):
+
+```bash
+./gradlew :pulsar-broker:test -PtestAsyncProfiler --tests "<SomeTest>" \
+ "-Ptest.asyncprofiler.opts=event=Java_java_lang_Throwable_fillInStackTrace,tree,reverse" \
+ -Ptest.asyncprofiler.outputformat=html
+```
+
+#### Profiling an integration-test cluster
+
+The module tests above profile the JVM the tests run in. To profile the **broker, bookies and
+ZooKeeper inside the containers** of a real cluster, run:
+
+```bash
+./gradlew :tests:integration:profilingIntegrationTest
+```
+
+That one task does everything the profiling run needs: it builds
+`apachepulsar/java-test-image:latest-asyncprofiler` (the test image with async-profiler installed —
+kept under its own tag so it never replaces the image the other integration tests use), relaxes the
+kernel `perf_event` limits that the `cpu` sampling engine needs by way of a privileged throwaway
+container, and runs
+[`PulsarProfilingTest`](tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingTest.java)
+against it with retries off. That test drives `pulsar-perf` against a single broker.
+
+**Any other integration test can be profiled without being modified**, by pointing the task at it and
+naming the cluster components to attach the profiler to:
+
+```bash
+./gradlew :tests:integration:profilingIntegrationTest --tests "<SomeIntegrationTest>" \
+ -Pinttest.asyncprofiler.components=broker,bookie
+```
+
+`components` takes `broker`, `proxy`, `functionworker`, `bookie`, `zookeeper`, or `all`, and defaults
+to `broker` for this task. It is what `PulsarClusterSpec.profileBroker` and its siblings fall back to,
+so it has no effect on a test that sets those flags itself — `PulsarProfilingTest` does, which is why
+it profiles the broker whatever you pass. Every other test leaves them at their default of off, so
+without this property nothing would be profiled. Setting it also enables manual tests, so
+`-Pinttest.asyncprofiler.components=<...>` profiles a cluster through the plain `integrationTest`
+task too.
+
+Recordings land in `tests/integration/build/`, named `inttest_profile_<commit>_<time>_<container>_<pid>.jfr`
+— the commit id comes from `git rev-parse --short HEAD` so profiles taken before and after a change
+can be told apart — one file per profiled container. Tune the run with
+`-Pinttest.asyncprofiler.opts=<agent options>` (default
+`event=cpu,lock=1ms,alloc=2m,jfrsync=profile`), `-Pinttest.asyncprofiler.outputformat=<ext>`,
+`-Pinttest.asyncprofiler.dir=<dir>` and `-Pgit.commit.id.abbrev=<id>`.
+
+Two flags worth knowing: `-Pinttest.asyncprofiler.skipPerfEventTuning` skips the privileged
+container, for a host where the `perf_event` values are already set through `sysctl` or where Docker
+disallows privileged containers (the run continues either way, with less accurate native stacks).
+`-Pdocker.wolfi` builds the base image from Wolfi, which is what makes the `GLIBC_TUNABLES` the test
+sets take effect.
+
+#### Analyzing a JFR file
+
+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.
+
### Integration tests
Integration tests live in `tests/` (see `tests/README.md`). They use
diff --git a/build-logic/conventions/src/main/kotlin/JfrFlamegraphsTask.kt b/build-logic/conventions/src/main/kotlin/JfrFlamegraphsTask.kt
new file mode 100644
index 000000000000..35ce67dcc5c5
--- /dev/null
+++ b/build-logic/conventions/src/main/kotlin/JfrFlamegraphsTask.kt
@@ -0,0 +1,132 @@
+/*
+ * 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.
+ */
+
+import org.gradle.api.DefaultTask
+import org.gradle.api.GradleException
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.provider.ListProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.process.ExecOperations
+import java.io.ByteArrayOutputStream
+import java.io.File
+import javax.inject.Inject
+
+/**
+ * Converts JFR recordings into flame graphs with async-profiler's `jfrconv`.
+ *
+ * Each recording gets a directory next to it named after the file without its extension plus a
+ * `-flamegraphs` suffix — `profile.jfr` produces `profile-flamegraphs/` — holding one flame graph
+ * per view. A view is a profiling event (`cpu`, `wall`, `alloc`, `lock`, …) rendered three ways:
+ * merged across threads (`cpu.html`), split per thread (`cpu_threads.html`) and grouped into
+ * async-profiler's categories (`cpu_classify.html`).
+ *
+ * A recording rarely contains every event: a run without allocation profiling has nothing to put in
+ * an `alloc` view. Such a view is skipped with a message rather than failing the task, so that one
+ * absent event does not cost the views that could be produced.
+ */
+abstract class JfrFlamegraphsTask : DefaultTask() {
+
+ /** The recordings to convert. */
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.NAME_ONLY)
+ abstract val jfrFiles: ConfigurableFileCollection
+
+ /** async-profiler's `jfrconv`. */
+ @get:Input
+ abstract val jfrconvExecutable: Property<String>
+
+ /** Profiling events to render, one directory entry per event and rendering. */
+ @get:Input
+ abstract val profileTypes: ListProperty<String>
+
+ @get:Inject
+ abstract val execOperations: ExecOperations
+
+ @TaskAction
+ fun convert() {
+ val recordings = jfrFiles.files.filter { it.isFile }.sortedBy { it.absolutePath }
+ if (recordings.isEmpty()) {
+ throw GradleException("No JFR recordings to convert. Point -Pjfr at a .jfr file or at a "
+ + "directory containing one.")
+ }
+ val jfrconv = jfrconvExecutable.get()
+ val types = profileTypes.get()
+ var produced = 0
+ var skipped = 0
+ for (recording in recordings) {
+ val baseName = recording.name.substringBeforeLast('.')
+ val outputDir = File(recording.parentFile, "$baseName-flamegraphs")
+ if (!outputDir.isDirectory && !outputDir.mkdirs()) {
+ throw GradleException("Cannot create flame graph directory ${outputDir.absolutePath}")
+ }
+ for (type in types) {
+ // Merged, per-thread and category-grouped renderings of the same event.
+ for ((suffix, extraArgs) in listOf(
+ "" to emptyList(),
+ "_threads" to listOf("--threads"),
+ "_classify" to listOf("--classify"),
+ )) {
+ val output = File(outputDir, "$type$suffix.html")
+ val title = "$baseName $type" + if (suffix.isEmpty()) "" else " (${suffix.trimStart('_')})"
+ if (runJfrconv(jfrconv, recording, output, type, extraArgs, title)) {
+ produced++
+ } else {
+ skipped++
+ }
+ }
+ }
+ logger.lifecycle("Flame graphs for {}: {}", recording.name, outputDir.absolutePath)
+ }
+ logger.lifecycle("Wrote {} flame graph(s) from {} recording(s){}", produced, recordings.size,
+ if (skipped > 0) ", skipped $skipped view(s) with no matching events" else "")
+ }
+
+ /** Returns true when the view was written, false when the recording holds no such event. */
+ private fun runJfrconv(jfrconv: String, recording: File, output: File, type: String,
+ extraArgs: List<String>, title: String): Boolean {
+ val stderr = ByteArrayOutputStream()
+ val result = execOperations.exec {
+ commandLine(buildList {
+ add(jfrconv)
+ add("--$type")
+ addAll(extraArgs)
+ add("--title")
+ add(title)
+ add(recording.absolutePath)
+ add(output.absolutePath)
+ })
+ errorOutput = stderr
+ standardOutput = ByteArrayOutputStream()
+ isIgnoreExitValue = true
+ }
+ if (result.exitValue == 0) {
+ return true
+ }
+ // jfrconv leaves a truncated file behind when it fails, which would look like a real view.
+ output.delete()
+ logger.info("Skipping the {} view of {}: {}", type, recording.name,
+ stderr.toString().trim().ifEmpty { "jfrconv exited with ${result.exitValue}" })
+ return false
+ }
+}
diff --git a/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts b/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts
index 75029e148569..a699bd2eaf4a 100644
--- a/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts
+++ b/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts
@@ -17,6 +17,8 @@
* under the License.
*/
+import java.io.File
+
plugins {
`java-library`
id("pulsar.code-quality-conventions")
@@ -279,6 +281,111 @@ tasks.withType<Test>().configureEach {
}
}
+// Run tests under async-profiler, enabled with the single property `-PtestAsyncProfiler`
+// (see CONTRIBUTING.md). The `test.asyncprofiler.*` properties below only tune the defaults and keep
+// the names of the `testAsyncProfiler` Maven profile that the 4.x branches use. This is a second
+// `configureEach` block so that it overrides the settings above, and so that the environment
+// variable and JDK lookups it does stay out of the configuration cache inputs when profiling is off.
+val asyncProfilerEnabled = providers.gradleProperty("testAsyncProfiler")
+ .map { it.isBlank() || it.toBoolean() }
+ .getOrElse(false)
+if (asyncProfilerEnabled) {
+ // Locate the agent library: an explicit -Ptest.asyncprofiler.libpath wins, then the
+ // LIBASYNCPROFILER_PATH environment variable (the variable microbench/README.md already uses for
+ // profiling JMH benchmarks), and finally the copy that Amazon Corretto ships inside the JDK that
+ // runs the tests.
+ val testJvmHome = testJavaVersion
+ .flatMap { version ->
+ javaToolchains.launcherFor { languageVersion.set(JavaLanguageVersion.of(version)) }
+ .map { it.metadata.installationPath.asFile.absolutePath }
+ }
+ .orElse(providers.systemProperty("java.home"))
+ val libraryPath = providers.gradleProperty("test.asyncprofiler.libpath")
+ .orElse(providers.environmentVariable("LIBASYNCPROFILER_PATH"))
+ .orElse(testJvmHome.map { jvmHome ->
+ sequenceOf("libasyncProfiler.so", "libasyncProfiler.dylib")
+ .map { File(jvmHome, "lib/$it") }
+ .firstOrNull(File::isFile)
+ ?.absolutePath
+ })
+ .orNull
+ if (libraryPath == null || !File(libraryPath).isFile) {
+ throw GradleException(
+ "-PtestAsyncProfiler is set but the async-profiler agent library "
+ + (libraryPath?.let { "'$it' does not exist" } ?: "could not be located")
+ + ". Set the LIBASYNCPROFILER_PATH environment variable (or "
+ + "-Ptest.asyncprofiler.libpath) to the full path of libasyncProfiler.so (Linux) or "
+ + "libasyncProfiler.dylib (macOS). Amazon Corretto ships one in \$JAVA_HOME/lib; "
+ + "otherwise install async-profiler from "
+ + "https://github.com/async-profiler/async-profiler/releases."
+ )
+ }
+ // itimer is the only CPU sampling engine available outside Linux; on Linux the perf_events based
+ // "cpu" engine produces the most accurate method profile.
+ // https://github.com/async-profiler/async-profiler/blob/master/docs/CpuSamplingEngines.md
+ val defaultEvent =
+ if (providers.systemProperty("os.name").get().startsWith("Linux")) "cpu" else "itimer"
+ val event = providers.gradleProperty("test.asyncprofiler.event").getOrElse(defaultEvent)
+ // "all" (async-profiler 4.1+) adds wall clock, allocation, lock and native profiling on top of
+ // the CPU engine that `event` selects.
+ val profilerOptions = providers.gradleProperty("test.asyncprofiler.opts")
+ .getOrElse("event=$event,all,alloc=2m,jfrsync=profile")
+ // async-profiler derives the output format from the file name extension (jfr, html, collapsed,
+ // folded), but only when nothing in `opts` already selects one — and the default `jfrsync` does.
+ // Changing this to html therefore also means overriding test.asyncprofiler.opts, otherwise the
+ // file is a JFR recording with an .html name.
+ val outputFormat = providers.gradleProperty("test.asyncprofiler.outputformat").getOrElse("jfr")
+ val outputDir = providers.gradleProperty("test.asyncprofiler.dir")
+ .map { rootProject.file(it) }
+ .getOrElse(rootProject.layout.buildDirectory.dir("test-profiles").get().asFile)
+
+ tasks.withType<Test>().configureEach {
+ val taskPath = path
+ // async-profiler expands %t (start timestamp) and %p (pid) itself, which keeps the profiles
+ // of separate runs, and of separate forks of one run, apart.
+ val baseName = "test_profile_" + taskPath.removePrefix(":").replace(':', '-')
+ val profileFile = File(outputDir, "${baseName}_%t_%p.$outputFormat")
+ val logFile = File(outputDir, "$baseName.log")
+
+ jvmArgs(
+ // DebugNonSafepoints is a diagnostic option. It makes the profiler's stack traces of
+ // JIT compiled frames accurate.
+ "-XX:+UnlockDiagnosticVMOptions",
+ "-XX:+DebugNonSafepoints",
+ "-agentpath:$libraryPath=start,$profilerOptions,quiet,file=$profileFile",
+ )
+ // Logging to the console distorts the profile, so send log4j2 output to a file next to it.
+ systemProperty("pulsar.test.logging.appender", "FILE")
+ systemProperty("pulsar.test.logging.file", logFile.absolutePath)
+ // The tests worth profiling are often the manual ones — long-running, load-generating cases
+ // that ManualTestUtil skips unless this is set. Profiling one is exactly the situation they
+ // exist for, so asking for a profile enables them. See ManualTestUtil.
+ systemProperty("pulsar.test.enableManualTest", "true")
+ // One test JVM at a time and no retries, so that a run produces a single comparable profile.
+ maxParallelForks = 1
+ systemProperty("testRetryCount", "0")
+ // A profiling run has to actually run the tests, even when the task is up-to-date. Don't
+ // "fix" this by declaring inputs: the point is to re-run, not to track a missing input. The
+ // profiler settings are not task inputs, so a profiling run would otherwise also store a
+ // build cache entry under the same key as an ordinary run.
+ outputs.upToDateWhen { false }
+ outputs.cacheIf("test runs under async-profiler are never cached") { false }
+ doFirst {
+ // async-profiler does not create the directory it writes the profile into.
+ outputDir.mkdirs()
+ val test = this as Test
+ // Pre-touch a fixed size heap so that heap growth and GC resizing don't distort the
+ // profile. This is read at execution time because a module may change maxHeapSize after
+ // this convention has run, and because Gradle folds a jvmArgs("-Xmx...") into it.
+ if (test.minHeapSize == null) {
+ test.minHeapSize = test.maxHeapSize
+ test.jvmArgs("-XX:+AlwaysPreTouch")
+ }
+ test.logger.lifecycle("Profiling {} with async-profiler into {}", taskPath, profileFile)
+ }
+ }
+}
+
// Expose test classes for cross-module test dependencies (Maven test-jar equivalent)
val testJar = tasks.register<Jar>("testJar") {
archiveClassifier.set("tests")
diff --git a/build.gradle.kts b/build.gradle.kts
index 55e0210f4b2c..6bd912c4a5f2 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -165,6 +165,53 @@ tasks.register("sanityCheck") {
}
// ── Parent POM publication ──────────────────────────────────────────────────
+// ── Flame graphs from JFR recordings ────────────────────────────────────────
+// Converts the recordings produced by -PtestAsyncProfiler or by
+// :tests:integration:profilingIntegrationTest into flame graphs. See CONTRIBUTING.md.
+// ./gradlew jfrFlamegraphs # every recording in build/test-profiles
+// ./gradlew jfrFlamegraphs -Pjfr=<file-or-dir> # a specific recording, or another directory
+tasks.register<JfrFlamegraphsTask>("jfrFlamegraphs") {
+ group = "verification"
+ description = "Convert JFR recordings into flame graphs next to them, in a <name>-flamegraphs " +
+ "directory. Pass -Pjfr=<file-or-directory> to pick what to convert."
+
+ val jfrInput = providers.gradleProperty("jfr")
+ .map { rootProject.file(it) }
+ .getOrElse(layout.buildDirectory.dir("test-profiles").get().asFile)
+ // A directory is wired in as a file tree rather than listed here: the listing has to happen when
+ // the task runs, or a reused configuration cache entry would convert the recordings of the
+ // previous profiling run and miss the one just made. The tree is searched recursively because
+ // JMH's async-profiler integration writes its recording into a directory per benchmark.
+ jfrFiles.from(if (jfrInput.isFile) jfrInput else fileTree(jfrInput) { include("**/*.jfr") })
+
+ // async-profiler's converter, looked up the same way as the profiler library itself:
+ // -Pjfrconv wins, then the bin/ directory of the async-profiler install that
+ // LIBASYNCPROFILER_PATH points into, then the copy that Amazon Corretto ships in the JDK that
+ // runs Gradle.
+ jfrconvExecutable.set(
+ providers.gradleProperty("jfrconv")
+ .orElse(providers.environmentVariable("LIBASYNCPROFILER_PATH")
+ .map { File(File(it).parentFile.parentFile, "bin/jfrconv").absolutePath })
+ .orElse(providers.systemProperty("java.home").map { File(it, "bin/jfrconv").absolutePath })
+ .map {
+ if (!File(it).canExecute()) {
+ throw GradleException("async-profiler's jfrconv was not found at '$it'. Set " +
+ "-Pjfrconv=<path>, or LIBASYNCPROFILER_PATH to the profiler library of an " +
+ "async-profiler install. Amazon Corretto ships jfrconv in the JDK's bin " +
+ "directory.")
+ }
+ it
+ })
+
+ // A recording collects several events; render every one that is present in it.
+ profileTypes.set(providers.gradleProperty("jfr.types")
+ .map { it.split(",").map(String::trim).filter(String::isNotEmpty) }
+ .getOrElse(listOf("cpu", "wall", "alloc", "lock")))
+
+ // Converting is cheap, and the recordings change on every profiling run.
+ outputs.upToDateWhen { false }
+}
+
// Publishes org.apache.pulsar:pulsar as a POM-only parent artifact.
// Child modules reference this via <parent> in their POMs, inheriting
// shared ASF metadata (license, SCM, organization, etc.).
diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/ManualTestUtil.java b/buildtools/src/main/java/org/apache/pulsar/tests/ManualTestUtil.java
index d64e550f5ca7..c71febee3194 100644
--- a/buildtools/src/main/java/org/apache/pulsar/tests/ManualTestUtil.java
+++ b/buildtools/src/main/java/org/apache/pulsar/tests/ManualTestUtil.java
@@ -23,12 +23,26 @@ import org.testng.SkipException;
public class ManualTestUtil {
+ /**
+ * System property equivalent of the {@code ENABLE_MANUAL_TEST} environment variable. The build
+ * sets it whenever a test task runs under a profiler, since the manual tests are usually the
+ * ones worth profiling, and a system property travels with the test JVM's command line where an
+ * environment variable depends on how the task was launched.
+ */
+ public static final String ENABLE_MANUAL_TEST_PROPERTY = "pulsar.test.enableManualTest";
+
public static void skipManualTestIfNotEnabled() {
- if (!BooleanUtils.toBoolean(System.getenv("ENABLE_MANUAL_TEST")) && !isRunningInIntelliJ()) {
- throw new SkipException("This test requires setting ENABLE_MANUAL_TEST=true environment variable.");
+ if (!isManualTestEnabled() && !isRunningInIntelliJ()) {
+ throw new SkipException("This test requires setting the " + ENABLE_MANUAL_TEST_PROPERTY
+ + " system property or the ENABLE_MANUAL_TEST=true environment variable.");
}
}
+ public static boolean isManualTestEnabled() {
+ return BooleanUtils.toBoolean(System.getProperty(ENABLE_MANUAL_TEST_PROPERTY))
+ || BooleanUtils.toBoolean(System.getenv("ENABLE_MANUAL_TEST"));
+ }
+
public static boolean isRunningInIntelliJ() {
// Check for IntelliJ-specific system properties
return System.getProperty("idea.test.cyclic.buffer.size") != null
diff --git a/buildtools/src/main/resources/log4j2.xml b/buildtools/src/main/resources/log4j2.xml
index f962d6198e1e..be3d0f939165 100644
--- a/buildtools/src/main/resources/log4j2.xml
+++ b/buildtools/src/main/resources/log4j2.xml
@@ -25,7 +25,7 @@
status="INFO">
<Properties>
<Property name="pulsar.test.logging.appender" value="CONSOLE"/>
- <Property name="pulsar.test.logging.file" value="target/pulsar.test.log"/>
+ <Property name="pulsar.test.logging.file" value="build/pulsar.test.log"/>
<Property name="pulsar.test.logging.file.immediateFlush" value="false"/>
</Properties>
<Appenders>
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 8f3b88b808f9..bb91d83c10d3 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -20,6 +20,7 @@
# Docker
docker-jdk = "25"
pulsar-client-python = "3.13.0"
+async-profiler = "4.5"
# Code quality
checkstyle = "13.3.0"
# Major frameworks
diff --git a/microbench/README.md b/microbench/README.md
index 6b1b8af593df..1704cdd9a9cb 100644
--- a/microbench/README.md
+++ b/microbench/README.md
@@ -23,6 +23,13 @@
This module contains microbenchmarks for Apache Pulsar.
+> **Run benchmarks on Linux x86_64 when the numbers matter.** That is Pulsar's most common deployment
+> target, and results from elsewhere do not carry over. `System.nanoTime()` is far more expensive on
+> macOS than on Linux, which skews the results in some cases — JMH's own measurement loop pays that
+> cost on every invocation. async-profiler also supports only some of its sampling engines on macOS,
+> so `-prof async` is less reliable there. Benchmarking on macOS or arm64 is fine while iterating —
+> just confirm the result on Linux x86_64 before drawing a conclusion from it.
+
## Running the benchmarks
The benchmarks are written using [JMH](http://openjdk.java.net/projects/code-tools/jmh/). To compile & run the benchmarks, use the following command:
@@ -58,7 +65,8 @@ java -jar microbench/build/libs/microbench-*-benchmarks.jar ".*BenchmarkName.*"
Running specific benchmarks with machine-readable output and saving the output to a file:
```shell
-java -jar microbench/build/libs/microbench-*-benchmarks.jar -rf json -rff jmh-result-$(date +%s).json ".*BenchmarkName.*" | tee jmh-result-$(date +%s).txt
+ts=$(date +%s)
+java -jar microbench/build/libs/microbench-*-benchmarks.jar -rf json -rff jmh-result-$ts.json ".*BenchmarkName.*" | tee jmh-result-$ts.txt
```
The `jmh-result-*.json` file can be used to visualize the results using [JMH Visualizer](https://jmh.morethan.io/).
@@ -90,8 +98,7 @@ LIBASYNCPROFILER_PATH=$HOME/async-profiler/lib/libasyncProfiler.dylib
Linux example:
```shell
-# macos example
-LIBASYNCPROFILER_PATH=$HOME/async-profiler/lib/libasyncProfiler.dylib
+LIBASYNCPROFILER_PATH=$HOME/async-profiler/lib/libasyncProfiler.so
```
Then run the benchmarks with the `-prof` argument:
@@ -99,10 +106,37 @@ Then run the benchmarks with the `-prof` argument:
java -jar microbench/build/libs/microbench-*-benchmarks.jar -prof async:libPath=$LIBASYNCPROFILER_PATH\;output=flamegraph\;dir=profile-results ".*BenchmarkName.*"
```
-When profiling on Mac OS, you might need to add `\;event=itimer` to the `-prof` argument since it's the only [async profiler CPU sampling engine that supports Mac OS](https://github.com/async-profiler/async-profiler/blob/master/docs/CpuSamplingEngines.md#summary). The default value for `event` is `cpu`.
+The default value for `event` is `cpu`, which is a request for the best available [CPU sampling engine](https://github.com/async-profiler/async-profiler/blob/master/docs/CpuSamplingEngines.md#summary) rather than a specific one, so what it resolves to depends on the platform. If the profiler fails to start, add `\;event=itimer` to the `-prof` argument: `itimer` is available everywhere.
It's possible to add options to the async-profiler that aren't supported by the JMH async-profiler plugin. This can be done by adding `rawCommand` option to the `-prof` argument. This example shows how to add `all` (new in Async Profiler 4.1), `jfrsync` (record JFR events such as garbage collection) and `cstack=vmx` options.
```shell
java -jar microbench/build/libs/microbench-*-benchmarks.jar -prof async:libPath=$LIBASYNCPROFILER_PATH\;output=jfr\;dir=profile-results\;rawCommand=all,jfrsync,cstack=vmx ".*BenchmarkName.*"
-```
\ No newline at end of file
+```
+
+Outside Linux this particular command needs `\;event=itimer` as well. `all` turns on wall clock
+profiling, and where the `cpu` engine falls back to the wall clock engine the profiler refuses to
+start with `Cannot start wall clock with the selected event`, which shows up as a `<failure>` on the
+first warmup iteration and an empty result directory.
+
+### Turning a JFR recording into flame graphs
+
+`output=jfr` writes one recording per benchmark, into a directory named after the benchmark under
+`dir=`. The `jfrFlamegraphs` Gradle task renders each recording into every view at once — point it at
+the whole output directory and it finds the recordings inside:
+
+```shell
+./gradlew jfrFlamegraphs -Pjfr=profile-results
+```
+
+Each recording gets a directory beside it named after the file without its extension plus a
+`-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.
+
+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
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/cache/BrokerEntryCacheRollingRestartTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/cache/BrokerEntryCacheRollingRestartTest.java
index 729e89666e89..a058dc092d97 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/cache/BrokerEntryCacheRollingRestartTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/cache/BrokerEntryCacheRollingRestartTest.java
@@ -407,12 +407,12 @@ public class BrokerEntryCacheRollingRestartTest extends AbstractBrokerEntryCache
// { cat pulsar-broker/target/rolling_restarts_result_header.txt; \
// cat pulsar-broker/target/rolling_restarts_result_*.csv } \
// | duckdb -c "select * from read_csv('/dev/stdin')" | cat
- File resultHeaderFile = new File("target/rolling_restarts_result_header.txt");
+ File resultHeaderFile = new File("build/rolling_restarts_result_header.txt");
String resultHeader =
"description\tproduced\tconsumed\tconsumers\tbk_reads\tbk_read_entries\thits\tmisses\trestarts\tts\n";
FileUtils.write(resultHeaderFile, resultHeader, StandardCharsets.UTF_8);
long ts = System.currentTimeMillis();
- File resultFile = new File("target/rolling_restarts_result_" + ts + ".csv");
+ File resultFile = new File("build/rolling_restarts_result_" + ts + ".csv");
List<String> csvColumns = new ArrayList<>();
Arrays.stream(msgArgs).map(Object::toString).forEach(csvColumns::add);
csvColumns.add(formatTimestampForCsv(ts));
diff --git a/tests/docker-images/java-test-image/Dockerfile b/tests/docker-images/java-test-image/Dockerfile
index f6e0d7a472f6..fe899835795d 100644
--- a/tests/docker-images/java-test-image/Dockerfile
+++ b/tests/docker-images/java-test-image/Dockerfile
@@ -44,13 +44,14 @@ COPY target/java-test-functions.jar /pulsar/examples/
COPY target/buildtools.jar /pulsar/lib/
ARG INSTALL_ASYNC_PROFILER=false
-# install async-profiler from https://github.com/async-profiler/async-profiler/releases/tag/v4.1 to /opt/async-profiler
+ARG ASYNC_PROFILER_VERSION=4.5
+# install async-profiler from https://github.com/async-profiler/async-profiler/releases to /opt/async-profiler
RUN if [ "${INSTALL_ASYNC_PROFILER}" = "true" ]; then \
mkdir -p /opt/async-profiler \
&& cd /opt/async-profiler \
&& curl -L \
- "https://github.com/async-profiler/async-profiler/releases/download/v4.1/async-profiler-4.1-linux-$([ "$(uname -m)" = "aarch64" ] && echo "arm64" || echo "x64").tar.gz" \
+ "https://github.com/async-profiler/async-profiler/releases/download/v${ASYNC_PROFILER_VERSION}/async-profiler-${ASYNC_PROFILER_VERSION}-linux-$([ "$(uname -m)" = "aarch64" ] && echo "arm64" || echo "x64").tar.gz" \
| tar -zxvf - --strip-components=1; \
else \
echo "async-profiler installation skipped"; \
- fi
\ No newline at end of file
+ fi
diff --git a/tests/docker-images/java-test-image/build.gradle.kts b/tests/docker-images/java-test-image/build.gradle.kts
index 80252ceaaedf..f50a1d1d6216 100644
--- a/tests/docker-images/java-test-image/build.gradle.kts
+++ b/tests/docker-images/java-test-image/build.gradle.kts
@@ -21,6 +21,7 @@ val pulsarVersion = project.version.toString()
val dockerOrganization = providers.gradleProperty("docker.organization").getOrElse("apachepulsar")
val dockerTag = providers.gradleProperty("docker.tag").getOrElse("latest")
val dockerPlatforms = providers.gradleProperty("docker.platforms").getOrElse("")
+val dockerInstallAsyncProfiler = providers.gradleProperty("docker.install.asyncprofiler").getOrElse("false")
// Ensure the parent project is configured before resolving cross-project task references.
// Required for --configure-on-demand: the Kotlin DSL needs parent ClassLoaderScopes to be locked.
@@ -78,28 +79,45 @@ val prepareBuildContext = tasks.register<Sync>("prepareBuildContext") {
into("${projectDir}/target")
}
-val dockerBuild = tasks.register<Exec>("dockerBuild") {
- group = "docker"
- description = "Build the java-test-image Docker image"
+fun registerDockerBuild(taskName: String, imageTag: String, installAsyncProfiler: String) =
+ tasks.register<Exec>(taskName) {
+ group = "docker"
+
+ dependsOn(":docker:pulsar-docker-image:dockerBuild", prepareBuildContext)
+
+ val imageName = "${dockerOrganization}/java-test-image:${imageTag}"
+ val pulsarImage = "${dockerOrganization}/pulsar:${dockerTag}"
+ val asyncProfilerVersion = libs.versions.async.profiler.get()
- dependsOn(":docker:pulsar-docker-image:dockerBuild", prepareBuildContext)
+ workingDir = projectDir
- val imageName = "${dockerOrganization}/java-test-image:${dockerTag}"
- val pulsarImage = "${dockerOrganization}/pulsar:${dockerTag}"
+ val args = mutableListOf(
+ "docker", "build",
+ "-t", imageName,
+ "--build-arg", "PULSAR_IMAGE=${pulsarImage}",
+ "--build-arg", "INSTALL_ASYNC_PROFILER=${installAsyncProfiler}",
+ "--build-arg", "ASYNC_PROFILER_VERSION=${asyncProfilerVersion}"
+ )
- workingDir = projectDir
+ if (dockerPlatforms.isNotEmpty()) {
+ args.addAll(listOf("--platform", dockerPlatforms))
+ }
- val args = mutableListOf(
- "docker", "build",
- "-t", imageName,
- "--build-arg", "PULSAR_IMAGE=${pulsarImage}",
- )
+ args.add(".")
- if (dockerPlatforms.isNotEmpty()) {
- args.addAll(listOf("--platform", dockerPlatforms))
+ commandLine(args)
}
- args.add(".")
+val dockerBuild = registerDockerBuild("dockerBuild", dockerTag, dockerInstallAsyncProfiler)
+dockerBuild.configure {
+ description = "Build the java-test-image Docker image"
+}
- commandLine(args)
+// A separate image so that a profiling run never replaces the image the other integration tests use,
+// and so that the async-profiler download stays out of the ordinary (and CI) test image build.
+// :tests:integration:profilingIntegrationTest builds and uses this one.
+val dockerBuildWithAsyncProfiler =
+ registerDockerBuild("dockerBuildWithAsyncProfiler", "${dockerTag}-asyncprofiler", "true")
+dockerBuildWithAsyncProfiler.configure {
+ description = "Build the java-test-image Docker image with async-profiler installed"
}
diff --git a/tests/docker-images/latest-version-image/Dockerfile b/tests/docker-images/latest-version-image/Dockerfile
index da56bb30b1d8..0c6988b61f6b 100644
--- a/tests/docker-images/latest-version-image/Dockerfile
+++ b/tests/docker-images/latest-version-image/Dockerfile
@@ -86,15 +86,16 @@ RUN mkdir -p pulsar
RUN chmod g+rwx pulsar
ARG INSTALL_ASYNC_PROFILER=false
-# install async-profiler from https://github.com/async-profiler/async-profiler/releases/tag/v4.1 to /opt/async-profiler
+ARG ASYNC_PROFILER_VERSION=4.5
+# install async-profiler from https://github.com/async-profiler/async-profiler/releases to /opt/async-profiler
RUN if [ "${INSTALL_ASYNC_PROFILER}" = "true" ]; then \
mkdir -p /opt/async-profiler \
&& cd /opt/async-profiler \
&& curl -L \
- "https://github.com/async-profiler/async-profiler/releases/download/v4.1/async-profiler-4.1-linux-$([ "$(uname -m)" = "aarch64" ] && echo "arm64" || echo "x64").tar.gz" \
+ "https://github.com/async-profiler/async-profiler/releases/download/v${ASYNC_PROFILER_VERSION}/async-profiler-${ASYNC_PROFILER_VERSION}-linux-$([ "$(uname -m)" = "aarch64" ] && echo "arm64" || echo "x64").tar.gz" \
| tar -zxvf - --strip-components=1; \
else \
echo "async-profiler installation skipped"; \
fi
-CMD bash
\ No newline at end of file
+CMD bash
diff --git a/tests/docker-images/latest-version-image/build.gradle.kts b/tests/docker-images/latest-version-image/build.gradle.kts
index 15510e0a7e84..eb8d2fc9463b 100644
--- a/tests/docker-images/latest-version-image/build.gradle.kts
+++ b/tests/docker-images/latest-version-image/build.gradle.kts
@@ -21,6 +21,7 @@ val pulsarVersion = project.version.toString()
val dockerOrganization = providers.gradleProperty("docker.organization").getOrElse("apachepulsar")
val dockerTag = providers.gradleProperty("docker.tag").getOrElse("latest")
val dockerPlatforms = providers.gradleProperty("docker.platforms").getOrElse("")
+val dockerInstallAsyncProfiler = providers.gradleProperty("docker.install.asyncprofiler").getOrElse("false")
val golangImage = providers.gradleProperty("docker.golang.image").getOrElse("golang:1.25-alpine")
// Ensure the parent project is configured before resolving cross-project task references.
@@ -107,6 +108,7 @@ val dockerBuild = tasks.register<Exec>("dockerBuild") {
val imageName = "${dockerOrganization}/pulsar-test-latest-version:${dockerTag}"
val pulsarImage = "${dockerOrganization}/pulsar:${dockerTag}"
+ val asyncProfilerVersion = libs.versions.async.profiler.get()
workingDir = projectDir
@@ -115,6 +117,8 @@ val dockerBuild = tasks.register<Exec>("dockerBuild") {
"-t", imageName,
"--build-arg", "PULSAR_IMAGE=${pulsarImage}",
"--build-arg", "GOLANG_IMAGE=${golangImage}",
+ "--build-arg", "INSTALL_ASYNC_PROFILER=${dockerInstallAsyncProfiler}",
+ "--build-arg", "ASYNC_PROFILER_VERSION=${asyncProfilerVersion}"
)
if (dockerPlatforms.isNotEmpty()) {
diff --git a/tests/integration/build.gradle.kts b/tests/integration/build.gradle.kts
index ac58e56f3c53..290a61d49ce2 100644
--- a/tests/integration/build.gradle.kts
+++ b/tests/integration/build.gradle.kts
@@ -93,6 +93,24 @@ val integrationTestSuiteFile = integrationTestSuiteFileProperty.getOrElse("pulsa
val integrationTestSuiteFileExplicit = integrationTestSuiteFileProperty.isPresent
val integrationTestGroups = providers.gradleProperty("testGroups").orNull
val integrationTestExcludedGroups = providers.gradleProperty("excludedTestGroups").orNull
+val integrationTestAsyncProfilerDir = providers.gradleProperty("inttest.asyncprofiler.dir")
+ .getOrElse(layout.buildDirectory.get().asFile.absolutePath)
+// Stamped into the async-profiler output file names so that profiles recorded from different
+// revisions can be told apart when they are compared. Resolved here rather than left to the caller:
+// a value nobody passes only produces empty segments in the file name. `git rev-parse` is
+// best-effort — a tree built without git, or without a .git directory, simply leaves the commit id
+// out of the name — and -Pgit.commit.id.abbrev overrides it. Nothing has to pass a timestamp in;
+// the profiler expands the run timestamp itself.
+val gitCommitIdAbbrev = providers.gradleProperty("git.commit.id.abbrev").orNull
+ ?: runCatching {
+ providers.exec {
+ commandLine("git", "rev-parse", "--short", "HEAD")
+ isIgnoreExitValue = true
+ }.standardOutput.asText.get().trim()
+ }.getOrDefault("")
+// Must match the image that :tests:java-test-image:dockerBuildWithAsyncProfiler tags.
+val dockerOrganization = providers.gradleProperty("docker.organization").getOrElse("apachepulsar")
+val dockerTag = providers.gradleProperty("docker.tag").getOrElse("latest")
val ideaActive = providers.systemProperty("idea.active").map { it.toBoolean() }.getOrElse(false)
// When `--tests` is passed on the CLI, let TestNG discover tests directly from the classpath
// instead of restricting discovery to the suite XML — unless -PintegrationTestSuiteFile was
@@ -100,10 +118,58 @@ val ideaActive = providers.systemProperty("idea.active").map { it.toBoolean() }.
val hasCliTestsFilter = gradle.startParameter.taskRequests
.flatMap { it.args }
.any { it == "--tests" }
-val integrationTest = tasks.register<Test>("integrationTest") {
+// Shared by `integrationTest` and `profilingIntegrationTest`: everything that is a property of
+// running this module's tests at all, rather than of how a particular task selects them.
+// `defaultProfiledComponents` is what the task profiles unless -Pinttest.asyncprofiler.components
+// says otherwise: nothing for `integrationTest`, the broker for the task that exists to profile.
+fun Test.configureIntegrationTestDefaults(defaultProfiledComponents: String = "") {
testClassesDirs = sourceSets.test.get().output.classesDirs
classpath = sourceSets.test.get().runtimeClasspath
+ systemProperty("currentVersion", project.version.toString())
+ systemProperty("buildDirectory", layout.buildDirectory.get().asFile.absolutePath)
+ systemProperty("inttest.asyncprofiler.dir", integrationTestAsyncProfilerDir)
+ providers.gradleProperty("inttest.asyncprofiler.opts").orNull?.let {
+ systemProperty("inttest.asyncprofiler.opts", it)
+ }
+ providers.gradleProperty("inttest.asyncprofiler.outputformat").orNull?.let {
+ systemProperty("inttest.asyncprofiler.outputformat", it)
+ }
+ systemProperty("git.commit.id.abbrev", gitCommitIdAbbrev)
+
+ // Cluster components to attach async-profiler to, which is also what decides whether this run
+ // counts as a profiling run.
+ val profiledComponents = providers.gradleProperty("inttest.asyncprofiler.components")
+ .getOrElse(defaultProfiledComponents)
+ systemProperty("inttest.asyncprofiler.components", profiledComponents)
+ if (profiledComponents.isNotEmpty()) {
+ // Profiling a cluster is exactly what the manual tests exist for, and asking for a profile
+ // only to have the test skip itself is a poor way to discover the flag. This is keyed on
+ // profiling rather than on the task, so it also covers `integrationTest` run with
+ // -Pinttest.asyncprofiler.components. See ManualTestUtil.
+ systemProperty("pulsar.test.enableManualTest", "true")
+ }
+
+ jvmArgs(
+ "-XX:+ExitOnOutOfMemoryError",
+ "-Xmx1G",
+ "-XX:MaxDirectMemorySize=1G",
+ )
+
+ maxParallelForks = 1
+ forkEvery = 0
+
+ testLogging {
+ events("passed", "skipped", "failed")
+ showExceptions = true
+ showStackTraces = true
+ exceptionFormat = org.gradle.api.tasks.testing.logging.TestExceptionFormat.FULL
+ }
+}
+
+val integrationTest = tasks.register<Test>("integrationTest") {
+ configureIntegrationTestDefaults()
+
if (!ideaActive && (!hasCliTestsFilter || integrationTestSuiteFileExplicit)) {
useTestNG {
suites("src/test/resources/${integrationTestSuiteFile}")
@@ -121,23 +187,90 @@ val integrationTest = tasks.register<Test>("integrationTest") {
val defaultTestRetryCount = if (ideaActive) "0" else "1"
systemProperty("testRetryCount", providers.gradleProperty("testRetryCount").getOrElse(defaultTestRetryCount))
systemProperty("testFailFast", failFastValue.toString())
+}
- systemProperty("currentVersion", project.version.toString())
- systemProperty("buildDirectory", layout.buildDirectory.get().asFile.absolutePath)
+// Profiling an integration test used to take a documented sequence of image builds, exported
+// environment variables and a privileged container run. `profilingIntegrationTest` below does all of
+// it, so profiling a cluster is one command. See CONTRIBUTING.md.
- jvmArgs(
- "-XX:+ExitOnOutOfMemoryError",
- "-Xmx1G",
- "-XX:MaxDirectMemorySize=1G",
+// The async-profiler `cpu` engine samples through perf_events, which the kernel gates. These are the
+// settings from https://github.com/async-profiler/async-profiler/blob/master/docs/Troubleshooting.md,
+// applied to the kernel that runs the containers (the Docker VM on macOS, the host on Linux) by a
+// privileged throwaway container. Skip it with -Pinttest.asyncprofiler.skipPerfEventTuning when the
+// values are already set, when the Docker setup disallows privileged containers, or when profiling
+// with an engine that does not need perf_events.
+val skipPerfEventTuning = providers.gradleProperty("inttest.asyncprofiler.skipPerfEventTuning").isPresent
+val tuneKernelPerfEvents = tasks.register<Exec>("tuneKernelPerfEvents") {
+ description = "Relax the kernel perf_event limits that async-profiler's cpu engine needs"
+ // Copied into a local: a task action that captured the script-level val would capture the
+ // script object with it, which the configuration cache cannot serialize.
+ val skip = skipPerfEventTuning
+ onlyIf { !skip }
+ // Same image as PulsarContainer.ALPINE_IMAGE_NAME
+ commandLine(
+ "docker", "run", "--rm", "--privileged",
+ "--cap-add", "SYS_ADMIN", "--security-opt", "seccomp=unconfined",
+ "alpine:3.24", "sh", "-c",
+ "echo 1 > /proc/sys/kernel/perf_event_paranoid "
+ + "&& echo 0 > /proc/sys/kernel/kptr_restrict "
+ + "&& echo 1024 > /proc/sys/kernel/perf_event_max_stack "
+ + "&& echo 2048 > /proc/sys/kernel/perf_event_mlock_kb "
+ // also optimize for -XX:+UseTransparentHugePages
+ + "&& echo madvise > /sys/kernel/mm/transparent_hugepage/enabled "
+ + "&& echo advise > /sys/kernel/mm/transparent_hugepage/shmem_enabled "
+ + "&& echo defer > /sys/kernel/mm/transparent_hugepage/defrag "
+ + "&& echo 1 > /sys/kernel/mm/transparent_hugepage/khugepaged/defrag"
)
+ // Best effort: profiling still works without it, only with less accurate native stacks, so a
+ // Docker setup that refuses privileged containers must not fail the whole profiling run.
+ isIgnoreExitValue = true
+ doLast {
+ if ((this as Exec).executionResult.get().exitValue != 0) {
+ logger.warn("Could not relax the kernel perf_event limits. Profiling continues, but the "
+ + "cpu engine may produce incomplete stacks. Set the values manually, or pass "
+ + "-Pinttest.asyncprofiler.skipPerfEventTuning to skip this step.")
+ }
+ }
+}
- maxParallelForks = 1
- forkEvery = 0
+val profilingTestClass = "org.apache.pulsar.tests.integration.profiling.PulsarProfilingTest"
+tasks.register<Test>("profilingIntegrationTest") {
+ group = "verification"
+ description = "Run $profilingTestClass with async-profiler enabled in the cluster containers. " +
+ "Pass --tests to profile a different integration test."
- testLogging {
- events("passed", "skipped", "failed")
- showExceptions = true
- showStackTraces = true
- exceptionFormat = org.gradle.api.tasks.testing.logging.TestExceptionFormat.FULL
+ // Unlike `integrationTest`, this task exists to profile, so it profiles the broker unless
+ // -Pinttest.asyncprofiler.components says otherwise. PulsarProfilingTest sets the spec flags
+ // itself and does not depend on this; every other integration test does, since they default off.
+ configureIntegrationTestDefaults(defaultProfiledComponents = "broker")
+
+ // Build the test image that carries the profiler, and prepare the kernel for it.
+ dependsOn(":tests:java-test-image:dockerBuildWithAsyncProfiler", tuneKernelPerfEvents)
+
+ // Target the test directly rather than through a suite XML, so that --tests can point this at
+ // any other integration test without a suite file having to exist for it.
+ if (!hasCliTestsFilter) {
+ filter {
+ includeTestsMatching(profilingTestClass)
+ }
+ }
+
+ environment("PULSAR_TEST_IMAGE_NAME",
+ "${dockerOrganization}/java-test-image:${dockerTag}-asyncprofiler")
+ // Leak detection is paranoid by default and would distort the allocation profile.
+ environment("NETTY_LEAK_DETECTION", "off")
+
+ // A retried test would profile the cluster twice into the same run.
+ systemProperty("testRetryCount", "0")
+ systemProperty("testFailFast", "true")
+ failFast = true
+
+ // A profiling run always has to run, and its result must never come from the build cache.
+ outputs.upToDateWhen { false }
+ outputs.cacheIf("profiling runs are never cached") { false }
+
+ val profileDir = integrationTestAsyncProfilerDir
+ doFirst {
+ logger.lifecycle("Profiling with async-profiler into {}", profileDir)
}
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
index a4be03103d8c..457222677d67 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java
@@ -93,6 +93,13 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
private final String httpPath;
@Setter
private boolean enableAsyncProfiler = false;
+ /**
+ * Directory the profiler writes into, letting a test keep its recordings apart from every other
+ * test's. Unset means the {@code inttest.asyncprofiler.dir} system property the build passes in,
+ * and failing that the working directory's {@code build}.
+ */
+ @Setter
+ private String profileDirectory;
public PulsarContainer(String clusterName,
String hostname,
@@ -316,7 +323,7 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
if (System.getProperty("integrationtest.coverage.dir") != null) {
coverageDirectory = new File(System.getProperty("integrationtest.coverage.dir"));
} else {
- coverageDirectory = new File("target");
+ coverageDirectory = new File("build");
}
if (!coverageDirectory.isDirectory()) {
@@ -367,22 +374,16 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
.withPrivileged(true);
});
- File asyncProfilerDir;
- if (isNotBlank(System.getProperty("inttest.asyncprofiler.dir"))) {
- asyncProfilerDir = new File(System.getProperty("inttest.asyncprofiler.dir"));
- } else {
- asyncProfilerDir = new File("target");
- }
+ File asyncProfilerDir = new File(resolveProfileDirectory());
if (!asyncProfilerDir.exists()) {
if (!asyncProfilerDir.mkdirs()) {
- throw new IllegalArgumentException(
- "inttest.asyncprofiler.dir '" + asyncProfilerDir.getAbsolutePath()
- + "' doesn't exist and cannot be created.");
+ throw new IllegalArgumentException("Profiler directory '" + asyncProfilerDir.getAbsolutePath()
+ + "' doesn't exist and cannot be created.");
}
}
if (!asyncProfilerDir.isDirectory()) {
throw new IllegalArgumentException(
- "inttest.asyncprofiler.dir '" + asyncProfilerDir.getAbsolutePath() + "' isn't a directory.");
+ "Profiler directory '" + asyncProfilerDir.getAbsolutePath() + "' isn't a directory.");
}
// change access to asyncProfilerDir to allow all access so the the container user can write to it
// This matters only on Linux
@@ -397,14 +398,36 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
StringBuilder sb = new StringBuilder();
sb.append("-agentpath:/opt/async-profiler/lib/libasyncProfiler.so=start,");
sb.append(System.getProperty("inttest.asyncprofiler.opts", "event=cpu,lock=1ms,alloc=2m,jfrsync=profile"));
- sb.append(",file=/profiles/inttest_profile_").append(System.getProperty("git.commit.id.abbrev", ""));
- sb.append("_").append(System.getProperty("build.timestamp", "").replace(' ', '_'));
- sb.append("_").append(getContainerName());
- sb.append("_").append("%p.").append(System.getProperty("inttest.asyncprofiler.outputformat", "jfr"));
+ StringBuilder fileName = new StringBuilder("inttest_profile");
+ // Set by the build; left out of the name when the revision could not be determined
+ String commitId = System.getProperty("git.commit.id.abbrev", "");
+ if (isNotBlank(commitId)) {
+ fileName.append('_').append(commitId);
+ }
+ // async-profiler expands %t (the time profiling started) and %p (the pid inside the
+ // container) itself, which is what keeps the profiles of separate runs apart
+ fileName.append("_%t_").append(getContainerName()).append("_%p.")
+ .append(System.getProperty("inttest.asyncprofiler.outputformat", "jfr"));
+ sb.append(",file=/profiles/").append(fileName);
initializePulsarExtraOpts();
appendToEnv("PULSAR_EXTRA_OPTS", "-XX:+UnlockDiagnosticVMOptions -XX:+DebugNonSafepoints " + sb);
}
+ /**
+ * Where the profiler writes, most specific first: the directory the test asked for through
+ * {@code PulsarClusterSpec.profileDirectory}, then the {@code inttest.asyncprofiler.dir} system
+ * property the build passes in, and finally {@code build} in the working directory.
+ *
+ * @return the directory to bind as /profiles in the container
+ */
+ private String resolveProfileDirectory() {
+ if (isNotBlank(profileDirectory)) {
+ return profileDirectory;
+ }
+ String fromBuild = System.getProperty("inttest.asyncprofiler.dir");
+ return isNotBlank(fromBuild) ? fromBuild : "build";
+ }
+
@Override
public boolean equals(Object o) {
if (!(o instanceof PulsarContainer)) {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingTest.java
index acfb4cb75367..bc1043ff48a1 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingTest.java
@@ -43,30 +43,20 @@ import org.testng.annotations.Test;
/**
* Sample test that profiles the broker side with Async Profiler.
*
- * Example usage:
- * # This has been tested on Mac with Orbstack (https://orbstack.dev/) docker
- * # compile integration test dependencies
- * ./gradlew assemble
- * # compile apachepulsar/java-test-image with async profiler (add "clean" to ensure a clean build with recent changes)
- * ./build/build_java_test_image.sh -Ddocker.install.asyncprofiler=true -Pdocker-wolfi
- * # set environment variables
- * export PULSAR_TEST_IMAGE_NAME=apachepulsar/java-test-image:latest
- * export NETTY_LEAK_DETECTION=off
- * export ENABLE_MANUAL_TEST=true
- * # enable perf events for profiling and tune it
- * docker run --rm -it --privileged --cap-add SYS_ADMIN --security-opt seccomp=unconfined \
- * alpine sh -c "echo 1 > /proc/sys/kernel/perf_event_paranoid \
- * && echo 0 > /proc/sys/kernel/kptr_restrict \
- * && echo 1024 > /proc/sys/kernel/perf_event_max_stack \
- * && echo 2048 > /proc/sys/kernel/perf_event_mlock_kb"
- * # translated to sysctl settings (for persistent configuration on Linux hosts)
+ * Example usage (this has been tested on Mac with Orbstack (https://orbstack.dev/) docker):
+ * ./gradlew :tests:integration:profilingIntegrationTest
+ * 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 this test against the result. See
+ * CONTRIBUTING.md for the properties that tune it.
+ * 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
* kernel.kptr_restrict=0
* kernel.perf_event_max_stack=1024
* kernel.perf_event_mlock_kb=2048
- * # run the test
- * ./gradlew :tests:integration:integrationTest -PintegrationTestSuiteFile=pulsar-profiling.xml -PtestRetryCount=0
- * By default, the .jfr files will go into tests/integration/build
+ * Add -Pdocker.wolfi to build the base image from Wolfi, which is what makes the GLIBC_TUNABLES
+ * below take effect.
+ * By default, the .jfr files and logs will go into tests/integration/build/pulsar-profiling
* You can use jfrconv from async profiler to convert them into html flamegraphs or use other tools such
* as Eclipse Mission Control (https://adoptium.net/jmc) or IntelliJ to open them.
*/
@@ -87,7 +77,8 @@ public class PulsarProfilingTest extends PulsarTestSuite {
private final String brokerHostname;
private final long numberOfMessages = 100_000_000;
- public PulsarPerfContainer(String clusterName,
+ public PulsarPerfContainer(File testOutputDir,
+ String clusterName,
String brokerHostname,
String hostname) {
super(PulsarContainer.DEFAULT_IMAGE_NAME);
@@ -96,27 +87,9 @@ public class PulsarProfilingTest extends PulsarTestSuite {
createContainerCmd.withHostName(hostname);
createContainerCmd.withName(clusterName + "-" + hostname);
});
- withEnv("PULSAR_MEM", DEFAULT_PULSAR_MEM);
+ withEnv("PULSAR_MEM", DEFAULT_PULSAR_MEM + " -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/testoutput");
withEnv("PULSAR_GC", "-XX:+UseZGC -XX:+ZGenerational");
setCommand("sleep 1000000");
- File testOutputDir = new File("target");
- if (!testOutputDir.exists()) {
- if (!testOutputDir.mkdirs()) {
- throw new IllegalArgumentException("Test output directory + '" + testOutputDir.getAbsolutePath()
- + "' doesn't exist and cannot be created.");
- }
- }
- if (!testOutputDir.isDirectory()) {
- throw new IllegalArgumentException(
- "Test output directory '" + testOutputDir.getAbsolutePath() + "' isn't a directory.");
- }
- // change access to testOutputDir to allow all access so the the container user can write to it
- // This matters only on Linux
- try {
- Files.setPosixFilePermissions(testOutputDir.toPath(), PosixFilePermissions.fromString("rwxrwxrwx"));
- } catch (IOException e) {
- throw new UncheckedIOException("Cannot change access to test output directory", e);
- }
withFileSystemBind(testOutputDir.getAbsolutePath(), "/testoutput", BindMode.READ_WRITE);
}
@@ -189,13 +162,36 @@ public class PulsarProfilingTest extends PulsarTestSuite {
private PulsarPerfContainer perfConsume;
private PulsarPerfContainer perfProduce;
private PulsarPerfContainer printStats;
+ private File testOutputDir;
@Override
public void setupCluster() throws Exception {
ManualTestUtil.skipManualTestIfNotEnabled();
+ createTestOutputDir();
super.setupCluster();
}
+ private void createTestOutputDir() {
+ testOutputDir = new File("build/pulsar-profiling");
+ if (!testOutputDir.exists()) {
+ if (!testOutputDir.mkdirs()) {
+ throw new IllegalArgumentException("Test output directory + '" + testOutputDir.getAbsolutePath()
+ + "' doesn't exist and cannot be created.");
+ }
+ }
+ if (!testOutputDir.isDirectory()) {
+ throw new IllegalArgumentException(
+ "Test output directory '" + testOutputDir.getAbsolutePath() + "' isn't a directory.");
+ }
+ // change access to testOutputDir to allow all access so the the container user can write to it
+ // This matters only on Linux
+ try {
+ Files.setPosixFilePermissions(testOutputDir.toPath(), PosixFilePermissions.fromString("rwxrwxrwx"));
+ } catch (IOException e) {
+ throw new UncheckedIOException("Cannot change access to test output directory", e);
+ }
+ }
+
@Override
public void tearDownCluster() throws Exception {
if (printStats != null) {
@@ -226,7 +222,7 @@ public class PulsarProfilingTest extends PulsarTestSuite {
protected void beforeStartCluster() throws Exception {
super.beforeStartCluster();
pulsarCluster.forEachContainer(
- // This is effective only when -Pdocker-wolfi has been passed when building java-test-image
+ // This is effective only when -Pdocker.wolfi has been passed when building java-test-image
// setting mmap_threshold explicitly will avoid it's dynamic increase
// https://sourceware.org/glibc/manual/latest/html_node/Memory-Allocation-Tunables.html
c -> c.withEnv("GLIBC_TUNABLES",
@@ -239,6 +235,7 @@ public class PulsarProfilingTest extends PulsarTestSuite {
// Enable profiling on the broker
specBuilder.profileBroker(true);
+ specBuilder.profileDirectory(testOutputDir.getAbsolutePath());
// Only run one broker so that all load goes to a single broker
specBuilder.numBrokers(1);
@@ -286,9 +283,9 @@ public class PulsarProfilingTest extends PulsarTestSuite {
// Create pulsar-perf containers
String brokerHostname = clusterName + "-pulsar-broker-0";
- perfProduce = new PulsarPerfContainer(clusterName, brokerHostname, "perf-produce");
- perfConsume = new PulsarPerfContainer(clusterName, brokerHostname, "perf-consume");
- printStats = new PulsarPerfContainer(clusterName, brokerHostname, "print-stats");
+ perfProduce = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "perf-produce");
+ perfConsume = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "perf-consume");
+ printStats = new PulsarPerfContainer(testOutputDir, clusterName, brokerHostname, "print-stats");
specBuilder.externalServices(Map.of(
"pulsar-produce", perfProduce,
"pulsar-consume", perfConsume,
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
index 1b3d5c8dc4ca..3856e1c21587 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
@@ -149,7 +149,7 @@ public class PulsarCluster {
.withEnv("pulsarNode", appendClusterName("pulsar-broker-0"));
metadataStoreUrl = appendClusterName(ZKContainer.NAME);
configurationMetadataStoreUrl = CSContainer.NAME + ":" + CS_PORT;
- zkContainer.setEnableAsyncProfiler(spec.profileZookeeper);
+ configureProfiling(zkContainer, spec.profileZookeeper);
}
this.csContainer = csContainer;
@@ -164,7 +164,7 @@ public class PulsarCluster {
.withEnv("metadataStoreUrl", metadataStoreUrl)
.withEnv("configurationMetadataStoreUrl", configurationMetadataStoreUrl)
.withEnv("clusterName", clusterName);
- proxyContainer.setEnableAsyncProfiler(spec.profileProxy);
+ configureProfiling(proxyContainer, spec.profileProxy);
// enable mTLS
if (spec.enableTls) {
@@ -231,7 +231,7 @@ public class PulsarCluster {
if (spec.bookieAdditionalPorts != null) {
spec.bookieAdditionalPorts.forEach(bookieContainer::addExposedPort);
}
- bookieContainer.setEnableAsyncProfiler(spec.profileBookie);
+ configureProfiling(bookieContainer, spec.profileBookie);
return bookieContainer;
})
);
@@ -280,7 +280,7 @@ public class PulsarCluster {
if (spec.brokerAdditionalPorts() != null) {
spec.brokerAdditionalPorts().forEach(brokerContainer::addExposedPort);
}
- brokerContainer.setEnableAsyncProfiler(spec.profileBroker);
+ configureProfiling(brokerContainer, spec.profileBroker);
return brokerContainer;
}
));
@@ -575,10 +575,22 @@ public class PulsarCluster {
.withEnv("zkServers", ZKContainer.NAME)
.withEnv(functionWorkerEnvs)
.withExposedPorts(functionWorkerAdditionalPorts.toArray(new Integer[0]));
- workerContainer.setEnableAsyncProfiler(spec.profileFunctionWorker);
+ configureProfiling(workerContainer, spec.profileFunctionWorker);
return workerContainer;
}
+ /**
+ * Applies the spec's profiling settings to a container. The output directory is set alongside
+ * the enable flag rather than at each call site, so that the two cannot drift apart.
+ *
+ * @param container the container to configure
+ * @param enabled whether this component is profiled
+ */
+ private void configureProfiling(PulsarContainer<?> container, boolean enabled) {
+ container.setEnableAsyncProfiler(enabled);
+ container.setProfileDirectory(spec.profileDirectory);
+ }
+
private void startFunctionWorkersWithThreadContainerFactory(String suffix, int numFunctionWorkers) {
workerContainers.putAll(runNumContainers(
"functions-worker-thread-" + suffix,
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java
index 7e6d272678fe..f678a7c98560 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.tests.integration.topologies;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@@ -200,18 +201,47 @@ public class PulsarClusterSpec {
@Default
boolean enableOxia = false;
+ // Which components async-profiler is attached to. A test that drives profiling itself, such as
+ // PulsarProfilingTest, sets these on the builder. Any other test can be profiled without being
+ // modified by naming the components on the command line instead — see
+ // inttest.asyncprofiler.components below.
@Default
- boolean profileBroker = false;
+ boolean profileBroker = isProfilingEnabledFor("broker");
@Default
- boolean profileProxy = false;
+ boolean profileProxy = isProfilingEnabledFor("proxy");
@Default
- boolean profileFunctionWorker = false;
+ boolean profileFunctionWorker = isProfilingEnabledFor("functionworker");
@Default
- boolean profileBookie = false;
+ boolean profileBookie = isProfilingEnabledFor("bookie");
@Default
- boolean profileZookeeper = false;
+ boolean profileZookeeper = isProfilingEnabledFor("zookeeper");
+
+ /**
+ * Directory the profiler writes its recordings into, so that a test can keep them apart from
+ * every other test's rather than having them all land in one place. Unset means the
+ * {@code inttest.asyncprofiler.dir} system property the build passes in, and failing that
+ * {@code build} in the working directory.
+ */
+ String profileDirectory;
+
+ /**
+ * Whether the given cluster component should be profiled with async-profiler, according to the
+ * {@code inttest.asyncprofiler.components} system property. The property holds a comma separated
+ * list of {@code broker}, {@code proxy}, {@code functionworker}, {@code bookie} and
+ * {@code zookeeper}, or {@code all} for every component. It is empty by default, so profiling
+ * stays off unless it is asked for.
+ *
+ * @param component the component name to look for
+ * @return true when the component is listed
+ */
+ private static boolean isProfilingEnabledFor(String component) {
+ String components = System.getProperty("inttest.asyncprofiler.components", "");
+ return Arrays.stream(components.split(","))
+ .map(String::trim)
+ .anyMatch(listed -> listed.equalsIgnoreCase(component) || listed.equalsIgnoreCase("all"));
+ }
}