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/20 20:04:48 UTC
(pulsar) 03/09: [improve][test] Add measurement-focused JFR profiling
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-improve-perf-measurement-jfr
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c12aa62d98e1de19451aa09cbc3b0ea987076ef9
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Sun Sep 20 14:24:06 2026 +0300
[improve][test] Add measurement-focused JFR profiling
---
tests/performance/README.md | 28 ++-
tests/performance/iot-telemetry.md | 10 +
tests/performance/launcher/build.gradle.kts | 7 +
.../pulsar/tests/performance/launcher/JfrCut.java | 228 +++++++++++++++++++++
.../launcher/JfrRecordingProcessor.java | 79 +++++++
.../performance/launcher/PerformanceLauncher.java | 38 +++-
.../tests/performance/launcher/JfrCutTest.java | 213 +++++++++++++++++++
.../scenarios/iot-telemetry-high-rate-profile.yaml | 2 +
.../scenarios/iot-telemetry-high-rate.yaml | 4 +-
tests/performance/scenarios/iot-telemetry.yaml | 2 +
.../tests/performance/tools/IotScenario.java | 15 +-
.../tests/performance/tools/TelemetryProducer.java | 27 ++-
.../tests/performance/tools/IotScenarioTest.java | 26 ++-
13 files changed, 672 insertions(+), 7 deletions(-)
diff --git a/tests/performance/README.md b/tests/performance/README.md
index 26739a6925a7..2e2bf84ef07b 100644
--- a/tests/performance/README.md
+++ b/tests/performance/README.md
@@ -81,7 +81,9 @@ file without interpreting unrelated sections. The launcher writes the fully reso
select its subtree with `--config-path`.
The `iotTelemetry` workload can run traffic before measurements begin. Use `warmupSeconds` with a positive `rate`,
-or use `warmupMessages` when `rate: 0`; the two settings are mutually exclusive. Warmup traffic remains part of
+or use `warmupMessages` when `rate: 0`; the two settings are mutually exclusive. `warmupRounds` repeats that
+traffic, and `warmupRoundDelaySeconds` adds an idle stabilization period after each fully drained round, including
+the final round. The default is one round with no delay. Warmup traffic remains part of
delivery and ordering validation. Producer throughput and the epoch-millisecond JFR measurement boundaries in
`producer-summary.json` cover only the configured measurement messages.
@@ -96,6 +98,8 @@ workloads:
profiling:
brokerOptions: event=cpu,interval=10ms,jfrsync=profile
producerOptions: ~
+ retainOriginalRecording: true
+ createMeasurementRecording: true
output:
directory: build/performance/iot-restart-profile
```
@@ -105,6 +109,28 @@ inherit other files recursively. Parents are applied in list order and the curre
merge recursively, while scalar values and lists replace earlier values. An explicit YAML `null` or `~` removes
an inherited entry. Cycles, missing files, non-mapping roots and invalid `extends` entries are rejected.
+Profiled standalone runs retain the complete JFR and also create a sibling whose name ends in
+`.measurement.jfr`. The measurement recording contains events that overlap the producer's recorded measurement
+interval, excluding startup and warmup. Set `profiling.retainOriginalRecording: false` to remove the complete
+recording after a successful cut, or `profiling.createMeasurementRecording: false` to keep only the complete
+recording. Both options default to `true` and apply to broker, producer and consumer recordings.
+
+Use the same cutter independently to select a different interval from an existing recording. `--from` and `--to`
+accept ISO-8601 instants, epoch milliseconds, or offsets from the first event such as `500ms`, `5s`, `2m`, `1h`,
+or `PT5S`. Omit `--from` to select from the beginning, or omit `--to` to select through the end. Use `--info`
+without either boundary to display the first and last event times and duration; it can also accompany a cut. The
+task requires JDK 19 or newer because it uses the public JFR recording writer added in that release:
+
+```bash
+./gradlew :tests:performance:launcher:runJfrCut \
+ --args='--input /tmp/full.jfr --from 5s --to 2m --output /tmp/measurement.jfr --info'
+```
+
+Java code can call `JfrCut.cut(Path input, Instant from, Instant to, Path output)` directly without invoking the
+command-line entry point. `JfrCut.cutUsingTimeExpressions(...)` provides the relative and omitted-boundary syntax,
+and `JfrCut.recordingInfo(...)` returns the event range. Events overlapping the half-open interval `[from, to)`
+are retained.
+
For one-off standalone overrides, prefix an existing scalar path with `PULSAR_PERFORMANCE_`, uppercase it and
separate path elements with underscores. The loader preserves the scalar's YAML type. For example:
diff --git a/tests/performance/iot-telemetry.md b/tests/performance/iot-telemetry.md
index def4317c1703..206b6dff0317 100644
--- a/tests/performance/iot-telemetry.md
+++ b/tests/performance/iot-telemetry.md
@@ -65,6 +65,10 @@ one key as required for Key_Shared delivery.
Warmup messages exercise the same producer, client, connection, topic and consumer paths as measured messages. They
remain in the monotonic device sequences and end-to-end delivery checks, but are excluded from throughput. For a
rate-limited workload, set `warmupSeconds`; for an unrestricted workload, set `warmupMessages`. Do not set both.
+The value applies to each of `warmupRounds`. Every round drains its asynchronous sends before
+`warmupRoundDelaySeconds` begins. The delay after the final round gives background JIT compilation and other
+startup work time to settle before the producer records the measurement boundary. This is a stabilization control,
+not a guarantee that the JVM has completed compilation.
`producer-summary.json` records the warmup and measurement counts and epoch-millisecond measurement boundaries so
the same window can be selected from broker and client JFRs.
@@ -98,6 +102,12 @@ The launcher owns each `file=` option so recordings remain inside the run direct
component unprofiled. The ordinary `run` task rejects profiling-enabled YAML rather than silently running with
an image that lacks the native agent.
+After every profiled process exits, the launcher uses the measurement boundaries from `producer-summary.json` to
+write a sibling `.measurement.jfr` containing events that overlap the measured interval. The complete recording
+is retained by default. Set `profiling.retainOriginalRecording: false` to keep only the measurement recording, or
+`profiling.createMeasurementRecording: false` to keep only the complete recording. If cutting fails, the complete
+recording is preserved even when its retention is disabled.
+
## Interpreting a run
A successful run sends the configured message count and reports the same number of unique messages for every
diff --git a/tests/performance/launcher/build.gradle.kts b/tests/performance/launcher/build.gradle.kts
index d74e278713a9..94887db4715f 100644
--- a/tests/performance/launcher/build.gradle.kts
+++ b/tests/performance/launcher/build.gradle.kts
@@ -61,3 +61,10 @@ tasks.register<JavaExec>("profile") {
outputs.upToDateWhen { false }
outputs.cacheIf("profiling runs are never cached") { false }
}
+
+tasks.register<JavaExec>("runJfrCut") {
+ group = "verification"
+ description = "Inspect or cut a JFR recording to an absolute or recording-relative time interval"
+ classpath = sourceSets.main.get().runtimeClasspath
+ mainClass.set("org.apache.pulsar.tests.performance.launcher.JfrCut")
+}
diff --git a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/JfrCut.java b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/JfrCut.java
new file mode 100644
index 000000000000..1d376923d499
--- /dev/null
+++ b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/JfrCut.java
@@ -0,0 +1,228 @@
+/*
+ * 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.performance.launcher;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.Callable;
+import java.util.function.Predicate;
+import jdk.jfr.consumer.RecordedEvent;
+import jdk.jfr.consumer.RecordingFile;
+import picocli.CommandLine;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
+
+/**
+ * Writes the events overlapping a time interval to a new JFR recording.
+ *
+ * <p>Call {@link #cut(Path, Instant, Instant, Path)} directly when embedding the cutter, or invoke this class as a
+ * command-line application through the {@code runJfrCut} Gradle task.
+ */
+@Command(name = "jfr-cut", mixinStandardHelpOptions = true,
+ description = "Write events overlapping a time interval to a new JFR recording")
+public final class JfrCut implements Callable<Integer> {
+ @Option(names = "--input", required = true, description = "Source JFR recording")
+ private Path input;
+
+ @Option(names = "--output", description = "Output JFR; defaults to <input>.cut.jfr")
+ private Path output;
+
+ @Option(names = {"--from", "--start"},
+ description = "Inclusive ISO-8601 instant, epoch milliseconds, or offset such as 5s")
+ private String from;
+
+ @Option(names = {"--to", "--end"},
+ description = "Exclusive ISO-8601 instant, epoch milliseconds, or offset such as 5s")
+ private String to;
+
+ @Option(names = "--info", description = "Show the recording event range and duration")
+ private boolean showInfo;
+
+ private JfrCut() {
+ }
+
+ public static void main(String[] args) {
+ System.exit(new CommandLine(new JfrCut()).execute(args));
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ RecordingInfo info = null;
+ if (showInfo) {
+ info = recordingInfo(input);
+ System.out.println("input=" + input.toAbsolutePath().normalize());
+ System.out.println("start=" + info.start());
+ System.out.println("end=" + info.end());
+ System.out.println("duration=" + info.duration());
+ System.out.println("durationMillis=" + info.duration().toMillis());
+ }
+ if (from == null && to == null && !showInfo) {
+ throw new IllegalArgumentException("Specify at least one JFR cut boundary");
+ }
+ if (from != null || to != null) {
+ cutUsingTimeExpressions(input, from, to, output != null ? output : defaultOutput(input), info);
+ }
+ return 0;
+ }
+
+ /**
+ * Resolves command-line time expressions and cuts a recording without invoking {@link #main(String[])}.
+ * Relative values such as {@code 5s} are measured from the first event in the recording. A {@code null} start
+ * or end selects the corresponding recording boundary.
+ */
+ public static void cutUsingTimeExpressions(Path input, String from, String to, Path output) throws IOException {
+ cutUsingTimeExpressions(input, from, to, output, null);
+ }
+
+ private static void cutUsingTimeExpressions(Path input, String from, String to, Path output,
+ RecordingInfo existingInfo) throws IOException {
+ RecordingInfo info = existingInfo != null ? existingInfo : recordingInfo(input);
+ Instant resolvedFrom = from != null ? parseTimeExpression(from, info.start()) : info.start();
+ Instant resolvedTo = to != null ? parseTimeExpression(to, info.start()) : info.exclusiveEnd();
+ cut(input, resolvedFrom, resolvedTo, output);
+ }
+
+ /**
+ * Writes events that overlap the half-open interval {@code [from, to)}.
+ *
+ * @param input source JFR recording
+ * @param from inclusive start of the selected interval
+ * @param to exclusive end of the selected interval
+ * @param output destination JFR recording, which must differ from {@code input}
+ */
+ public static void cut(Path input, Instant from, Instant to, Path output) throws IOException {
+ if (!from.isBefore(to)) {
+ throw new IllegalArgumentException("JFR cut start must be before its end");
+ }
+ Path normalizedInput = input.toAbsolutePath().normalize();
+ Path normalizedOutput = output.toAbsolutePath().normalize();
+ if (normalizedInput.equals(normalizedOutput)) {
+ throw new IllegalArgumentException("JFR cut output must differ from its input");
+ }
+
+ Path temporary = normalizedOutput.resolveSibling(normalizedOutput.getFileName() + ".tmp");
+ Files.deleteIfExists(temporary);
+ try (RecordingFile recording = new RecordingFile(normalizedInput)) {
+ write(recording, temporary, event -> event.getStartTime().isBefore(to)
+ && !event.getEndTime().isBefore(from));
+ }
+ try {
+ Files.move(temporary, normalizedOutput, StandardCopyOption.ATOMIC_MOVE,
+ StandardCopyOption.REPLACE_EXISTING);
+ } catch (java.nio.file.AtomicMoveNotSupportedException unsupported) {
+ Files.move(temporary, normalizedOutput, StandardCopyOption.REPLACE_EXISTING);
+ } finally {
+ Files.deleteIfExists(temporary);
+ }
+ }
+
+ static Instant parseInstant(String value) {
+ try {
+ return Instant.ofEpochMilli(Long.parseLong(value));
+ } catch (NumberFormatException notEpochMilliseconds) {
+ return Instant.parse(value);
+ }
+ }
+
+ static Instant parseTimeExpression(String value, Instant recordingStart) {
+ if (value.matches("[0-9]+(ms|s|m|h)")) {
+ int unitStart = 0;
+ while (unitStart < value.length() && Character.isDigit(value.charAt(unitStart))) {
+ unitStart++;
+ }
+ long amount = Long.parseLong(value.substring(0, unitStart));
+ Duration offset = switch (value.substring(unitStart)) {
+ case "ms" -> Duration.ofMillis(amount);
+ case "s" -> Duration.ofSeconds(amount);
+ case "m" -> Duration.ofMinutes(amount);
+ case "h" -> Duration.ofHours(amount);
+ default -> throw new IllegalArgumentException("Unsupported relative time unit in " + value);
+ };
+ return recordingStart.plus(offset);
+ }
+ if (value.startsWith("P")) {
+ return recordingStart.plus(Duration.parse(value));
+ }
+ return parseInstant(value);
+ }
+
+ static Path defaultOutput(Path input) {
+ String name = input.getFileName().toString();
+ String basename = name.endsWith(".jfr") ? name.substring(0, name.length() - 4) : name;
+ return input.resolveSibling(basename + ".cut.jfr");
+ }
+
+ /** Returns the first and last event times and their duration. */
+ public static RecordingInfo recordingInfo(Path input) throws IOException {
+ Instant start = null;
+ Instant end = null;
+ try (RecordingFile recording = new RecordingFile(input.toAbsolutePath().normalize())) {
+ while (recording.hasMoreEvents()) {
+ RecordedEvent event = recording.readEvent();
+ if (start == null || event.getStartTime().isBefore(start)) {
+ start = event.getStartTime();
+ }
+ if (end == null || event.getEndTime().isAfter(end)) {
+ end = event.getEndTime();
+ }
+ }
+ }
+ if (start == null) {
+ throw new IOException("Cannot cut a JFR recording with no events: " + input);
+ }
+ return new RecordingInfo(start, end, Duration.between(start, end));
+ }
+
+ /** Event time range found in a JFR recording. */
+ public record RecordingInfo(Instant start, Instant end, Duration duration) {
+ Instant exclusiveEnd() {
+ return end.equals(Instant.MAX) ? end : end.plusNanos(1);
+ }
+ }
+
+ private static void write(RecordingFile recording, Path output, Predicate<RecordedEvent> filter)
+ throws IOException {
+ final Method writeMethod;
+ try {
+ writeMethod = RecordingFile.class.getMethod("write", Path.class, Predicate.class);
+ } catch (NoSuchMethodException error) {
+ throw new IOException("Cutting JFR recordings requires JDK 19 or newer", error);
+ }
+ try {
+ writeMethod.invoke(recording, output, filter);
+ } catch (IllegalAccessException error) {
+ throw new IOException("Cannot access the public JFR recording writer", error);
+ } catch (InvocationTargetException error) {
+ Throwable cause = error.getCause();
+ if (cause instanceof IOException ioException) {
+ throw ioException;
+ }
+ if (cause instanceof RuntimeException runtimeException) {
+ throw runtimeException;
+ }
+ throw new IOException("JFR recording writer failed", cause);
+ }
+ }
+}
diff --git a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/JfrRecordingProcessor.java b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/JfrRecordingProcessor.java
new file mode 100644
index 000000000000..856689964674
--- /dev/null
+++ b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/JfrRecordingProcessor.java
@@ -0,0 +1,79 @@
+/*
+ * 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.performance.launcher;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+final class JfrRecordingProcessor {
+ private static final String JFR_SUFFIX = ".jfr";
+ private static final String MEASUREMENT_SUFFIX = ".measurement.jfr";
+
+ private JfrRecordingProcessor() {
+ }
+
+ static Set<Path> findOriginalRecordings(Path directory) throws IOException {
+ if (!Files.isDirectory(directory)) {
+ return Set.of();
+ }
+ try (Stream<Path> files = Files.walk(directory)) {
+ return files.filter(Files::isRegularFile)
+ .filter(path -> path.getFileName().toString().endsWith(JFR_SUFFIX))
+ .filter(path -> !path.getFileName().toString().endsWith(MEASUREMENT_SUFFIX))
+ .map(path -> path.toAbsolutePath().normalize())
+ .sorted()
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+ }
+
+ static void process(Set<Path> recordings, Instant from, Instant to,
+ boolean retainOriginal, boolean createMeasurementRecording) throws IOException {
+ IOException failure = null;
+ for (Path recording : recordings) {
+ try {
+ if (createMeasurementRecording) {
+ JfrCut.cut(recording, from, to, measurementPath(recording));
+ }
+ if (!retainOriginal) {
+ Files.delete(recording);
+ }
+ } catch (IOException error) {
+ if (failure == null) {
+ failure = new IOException("Failed to process JFR recordings");
+ }
+ failure.addSuppressed(error);
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ static Path measurementPath(Path recording) {
+ String name = recording.getFileName().toString();
+ return recording.resolveSibling(name.substring(0, name.length() - JFR_SUFFIX.length())
+ + MEASUREMENT_SUFFIX);
+ }
+}
diff --git a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/PerformanceLauncher.java b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/PerformanceLauncher.java
index 62cabeb3ac04..2517f0fdb9e8 100644
--- a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/PerformanceLauncher.java
+++ b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/PerformanceLauncher.java
@@ -26,9 +26,11 @@ import java.io.DataInputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.tests.integration.containers.PulsarContainer;
@@ -73,6 +75,8 @@ public class PerformanceLauncher implements Callable<Integer> {
String brokerProfileOptions = text(profiling, "brokerOptions");
String producerProfileOptions = text(profiling, "producerOptions");
String consumerProfileOptions = text(profiling, "consumerOptions");
+ boolean retainOriginalRecording = booleanValue(profiling, "retainOriginalRecording", true);
+ boolean createMeasurementRecording = booleanValue(profiling, "createMeasurementRecording", true);
boolean profilingEnabled = brokerProfileOptions != null || producerProfileOptions != null
|| consumerProfileOptions != null;
if (profilingEnabled
@@ -88,6 +92,7 @@ public class PerformanceLauncher implements Callable<Integer> {
: Path.of(loader.select(resolved, "output.directory").textValue());
runOutput = runOutput.toAbsolutePath().normalize();
Files.createDirectories(runOutput);
+ Set<Path> recordingsBeforeRun = JfrRecordingProcessor.findOriginalRecordings(runOutput);
Path brokerProfileDirectory = runOutput.resolve("broker-profile");
if (brokerProfileOptions != null) {
Files.createDirectories(brokerProfileDirectory);
@@ -156,7 +161,6 @@ public class PerformanceLauncher implements Callable<Integer> {
}
}
verifyStates(runOutput, applications);
- return 0;
} finally {
if (producer != null) {
saveContainerLog(producer, runOutput.resolve("producer/container.log"));
@@ -169,6 +173,19 @@ public class PerformanceLauncher implements Callable<Integer> {
}
cluster.stop();
}
+ if (profilingEnabled) {
+ JsonNode summary = loader.mapper().readTree(runOutput.resolve("producer/producer-summary.json").toFile());
+ Instant measurementStart = Instant.ofEpochMilli(requiredLong(summary, "measurementStartEpochMs"));
+ Instant measurementEnd = Instant.ofEpochMilli(requiredLong(summary, "measurementEndEpochMs"));
+ Set<Path> recordings = JfrRecordingProcessor.findOriginalRecordings(runOutput);
+ recordings.removeAll(recordingsBeforeRun);
+ if (recordings.isEmpty()) {
+ throw new IllegalStateException("Profiling completed without producing a JFR recording");
+ }
+ JfrRecordingProcessor.process(recordings, measurementStart, measurementEnd,
+ retainOriginalRecording, createMeasurementRecording);
+ }
+ return 0;
}
private GenericContainer<?> workloadContainer(PulsarCluster cluster, Path tools, Path configFile,
@@ -212,6 +229,25 @@ public class PerformanceLauncher implements Callable<Integer> {
return value.isTextual() && !value.textValue().isBlank() ? value.textValue() : null;
}
+ private static boolean booleanValue(JsonNode parent, String field, boolean defaultValue) {
+ JsonNode value = parent.path(field);
+ if (value.isMissingNode() || value.isNull()) {
+ return defaultValue;
+ }
+ if (!value.isBoolean()) {
+ throw new IllegalArgumentException("profiling." + field + " must be a boolean");
+ }
+ return value.booleanValue();
+ }
+
+ private static long requiredLong(JsonNode parent, String field) {
+ JsonNode value = parent.path(field);
+ if (!value.canConvertToLong()) {
+ throw new IllegalArgumentException("Missing numeric producer summary field " + field);
+ }
+ return value.longValue();
+ }
+
private static int waitForExit(GenericContainer<?> container, int timeoutSeconds) throws Exception {
return container.getDockerClient().waitContainerCmd(container.getContainerId()).start()
.awaitStatusCode(timeoutSeconds, TimeUnit.SECONDS);
diff --git a/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/JfrCutTest.java b/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/JfrCutTest.java
new file mode 100644
index 000000000000..a4b2f523f339
--- /dev/null
+++ b/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/JfrCutTest.java
@@ -0,0 +1,213 @@
+/*
+ * 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.performance.launcher;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.Set;
+import jdk.jfr.Event;
+import jdk.jfr.Name;
+import jdk.jfr.Recording;
+import jdk.jfr.consumer.RecordingFile;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class JfrCutTest {
+ private static final String EVENT_NAME = "org.apache.pulsar.tests.performance.Marker";
+
+ @Name(EVENT_NAME)
+ static class MarkerEvent extends Event {
+ String marker;
+ }
+
+ @Test
+ public void cutsEventsToMeasurementInterval() throws Exception {
+ Path directory = Files.createTempDirectory("jfr-cut-test");
+ try {
+ Path input = directory.resolve("input.jfr");
+ Instant[] interval = createRecording(input);
+ Path output = directory.resolve("output.jfr");
+
+ JfrCut.cut(input, interval[0], interval[1], output);
+
+ assertEquals(markers(output), List.of("measurement"));
+ assertEquals(markers(input), List.of("before", "measurement", "after"));
+ } finally {
+ deleteDirectory(directory);
+ }
+ }
+
+ @Test(dataProvider = "retentionModes")
+ public void appliesIndependentRetentionOptions(boolean retainOriginal, boolean createMeasurement)
+ throws Exception {
+ Path directory = Files.createTempDirectory("jfr-retention-test");
+ try {
+ Path input = directory.resolve("profile.jfr");
+ createRecording(input);
+ Path measurement = JfrRecordingProcessor.measurementPath(input);
+
+ JfrRecordingProcessor.process(Set.of(input), Instant.EPOCH,
+ Instant.now().plus(1, ChronoUnit.HOURS), retainOriginal, createMeasurement);
+
+ assertEquals(Files.exists(input), retainOriginal);
+ assertEquals(Files.exists(measurement), createMeasurement);
+ } finally {
+ deleteDirectory(directory);
+ }
+ }
+
+ @Test
+ public void findsOnlyOriginalRecordings() throws Exception {
+ Path directory = Files.createTempDirectory("jfr-discovery-test");
+ try {
+ Path original = directory.resolve("profile.jfr");
+ Path measurement = directory.resolve("profile.measurement.jfr");
+ Files.write(original, new byte[] {1});
+ Files.write(measurement, new byte[] {1});
+ Files.writeString(directory.resolve("profile.log"), "log");
+
+ assertEquals(JfrRecordingProcessor.findOriginalRecordings(directory), Set.of(original));
+ } finally {
+ deleteDirectory(directory);
+ }
+ }
+
+ @Test
+ public void rejectsInvalidInterval() throws Exception {
+ Path directory = Files.createTempDirectory("jfr-invalid-interval-test");
+ try {
+ Path input = directory.resolve("input.jfr");
+ Files.write(input, new byte[] {1});
+ Instant now = Instant.now();
+ try {
+ JfrCut.cut(input, now, now, directory.resolve("output.jfr"));
+ } catch (IllegalArgumentException expected) {
+ assertTrue(expected.getMessage().contains("start must be before"));
+ return;
+ }
+ fail("Expected an invalid interval to be rejected");
+ } finally {
+ deleteDirectory(directory);
+ }
+ }
+
+ @Test
+ public void parsesIsoInstantsAndEpochMilliseconds() {
+ Instant instant = Instant.parse("2026-09-20T10:05:00Z");
+
+ assertEquals(JfrCut.parseInstant(instant.toString()), instant);
+ assertEquals(JfrCut.parseInstant(Long.toString(instant.toEpochMilli())), instant);
+ }
+
+ @Test
+ public void parsesRecordingRelativeTimes() {
+ Instant start = Instant.parse("2026-09-20T10:00:00Z");
+
+ assertEquals(JfrCut.parseTimeExpression("500ms", start), start.plusMillis(500));
+ assertEquals(JfrCut.parseTimeExpression("5s", start), start.plusSeconds(5));
+ assertEquals(JfrCut.parseTimeExpression("2m", start), start.plus(2, ChronoUnit.MINUTES));
+ assertEquals(JfrCut.parseTimeExpression("1h", start), start.plus(1, ChronoUnit.HOURS));
+ assertEquals(JfrCut.parseTimeExpression("PT5S", start), start.plusSeconds(5));
+ }
+
+ @Test
+ public void supportsOmittedTimeBoundaries() throws Exception {
+ Path directory = Files.createTempDirectory("jfr-relative-cut-test");
+ try {
+ Path input = directory.resolve("input.jfr");
+ createRecording(input);
+ Path fromBeginning = directory.resolve("from-beginning.jfr");
+ Path throughEnd = directory.resolve("through-end.jfr");
+
+ JfrCut.cutUsingTimeExpressions(input, null, "1h", fromBeginning);
+ JfrCut.cutUsingTimeExpressions(input, "0ms", null, throughEnd);
+
+ assertEquals(markers(fromBeginning), List.of("before", "measurement", "after"));
+ assertEquals(markers(throughEnd), List.of("before", "measurement", "after"));
+ JfrCut.RecordingInfo info = JfrCut.recordingInfo(input);
+ assertTrue(!info.end().isBefore(info.start()));
+ assertEquals(info.duration(), Duration.between(info.start(), info.end()));
+ } finally {
+ deleteDirectory(directory);
+ }
+ }
+
+ @Test
+ public void derivesDefaultOutputBesideInput() {
+ assertEquals(JfrCut.defaultOutput(Path.of("/tmp/profile.jfr")), Path.of("/tmp/profile.cut.jfr"));
+ assertEquals(JfrCut.defaultOutput(Path.of("/tmp/profile")), Path.of("/tmp/profile.cut.jfr"));
+ }
+
+ @DataProvider
+ public Object[][] retentionModes() {
+ return new Object[][] {
+ {true, true},
+ {true, false},
+ {false, true},
+ {false, false}
+ };
+ }
+
+ private static Instant[] createRecording(Path path) throws Exception {
+ try (Recording recording = new Recording()) {
+ recording.enable(MarkerEvent.class);
+ recording.start();
+ marker("before");
+ Thread.sleep(10);
+ Instant from = Instant.now();
+ Thread.sleep(10);
+ marker("measurement");
+ Thread.sleep(10);
+ Instant to = Instant.now();
+ Thread.sleep(10);
+ marker("after");
+ recording.stop();
+ recording.dump(path);
+ return new Instant[] {from, to};
+ }
+ }
+
+ private static void marker(String marker) {
+ MarkerEvent event = new MarkerEvent();
+ event.marker = marker;
+ event.commit();
+ }
+
+ private static List<String> markers(Path path) throws Exception {
+ return RecordingFile.readAllEvents(path).stream()
+ .filter(event -> event.getEventType().getName().equals(EVENT_NAME))
+ .map(event -> event.getString("marker"))
+ .toList();
+ }
+
+ private static void deleteDirectory(Path directory) throws Exception {
+ try (var paths = Files.walk(directory)) {
+ for (Path path : paths.sorted(java.util.Comparator.reverseOrder()).toList()) {
+ Files.deleteIfExists(path);
+ }
+ }
+ }
+}
diff --git a/tests/performance/scenarios/iot-telemetry-high-rate-profile.yaml b/tests/performance/scenarios/iot-telemetry-high-rate-profile.yaml
index 70513d551c9f..07b84ca5b12d 100644
--- a/tests/performance/scenarios/iot-telemetry-high-rate-profile.yaml
+++ b/tests/performance/scenarios/iot-telemetry-high-rate-profile.yaml
@@ -22,5 +22,7 @@ profiling:
brokerOptions: event=cpu,interval=10ms,lock=1ms,alloc=2m,jfrsync=profile
producerOptions: event=cpu,interval=10ms,lock=1ms,alloc=2m,jfrsync=profile
consumerOptions: ""
+ retainOriginalRecording: true
+ createMeasurementRecording: true
output:
directory: build/performance/iot-telemetry-high-rate-profile
diff --git a/tests/performance/scenarios/iot-telemetry-high-rate.yaml b/tests/performance/scenarios/iot-telemetry-high-rate.yaml
index 8d6e492ae5b0..b43360f4a37a 100644
--- a/tests/performance/scenarios/iot-telemetry-high-rate.yaml
+++ b/tests/performance/scenarios/iot-telemetry-high-rate.yaml
@@ -28,7 +28,9 @@ workloads:
iotTelemetry:
rate: 0
warmupSeconds: 0
- warmupMessages: 1000000
+ warmupMessages: 500000
+ warmupRounds: 2
+ warmupRoundDelaySeconds: 5
numberOfMessages: 5000000
deviceCount: 1000000
gatewayCount: 500
diff --git a/tests/performance/scenarios/iot-telemetry.yaml b/tests/performance/scenarios/iot-telemetry.yaml
index 3b76f69d772a..035cf6ed9ae7 100644
--- a/tests/performance/scenarios/iot-telemetry.yaml
+++ b/tests/performance/scenarios/iot-telemetry.yaml
@@ -48,6 +48,8 @@ workloads:
# Warmup traffic is validated but excluded from throughput and the exported measurement window.
warmupSeconds: 20
warmupMessages: 0
+ warmupRounds: 1
+ warmupRoundDelaySeconds: 0
rate: 1000
numberOfMessages: 0
payloadBytes: 64
diff --git a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/IotScenario.java b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/IotScenario.java
index c0514ceb5695..d047ea0222ce 100644
--- a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/IotScenario.java
+++ b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/IotScenario.java
@@ -24,6 +24,7 @@ import java.util.List;
/** Configuration selected from the {@code workloads.iotTelemetry} scenario subtree. */
public record IotScenario(String serviceUrl, String topicPrefix, String subscriptionPrefix,
int durationSeconds, int warmupSeconds, long warmupMessages,
+ int warmupRounds, int warmupRoundDelaySeconds,
int rate, long numberOfMessages, int payloadBytes, int deviceCount,
int gatewayCount, int topicCount, int applicationCount,
int clientsPerApplication, int ioThreads, int listenerThreads,
@@ -31,18 +32,24 @@ public record IotScenario(String serviceUrl, String topicPrefix, String subscrip
int consumerTimeoutSeconds,
int clientRestartIntervalSeconds, double clientRestartFraction) {
public IotScenario {
+ // Preserve compatibility with configurations written before warmup rounds were introduced.
+ warmupRounds = warmupRounds == 0 ? 1 : warmupRounds;
if (serviceUrl == null || serviceUrl.isBlank() || topicPrefix == null || topicPrefix.isBlank()
|| subscriptionPrefix == null || subscriptionPrefix.isBlank()) {
throw new IllegalArgumentException("Service URL, topic prefix and subscription prefix are required");
}
- if (durationSeconds < 1 || warmupSeconds < 0 || warmupMessages < 0
+ long minimumRuntimeSeconds = Math.addExact(durationSeconds,
+ Math.addExact(Math.multiplyExact((long) warmupSeconds, warmupRounds),
+ Math.multiplyExact((long) warmupRoundDelaySeconds, warmupRounds)));
+ if (durationSeconds < 1 || warmupSeconds < 0 || warmupMessages < 0 || warmupRounds < 1
+ || warmupRoundDelaySeconds < 0
|| (warmupSeconds > 0 && warmupMessages > 0)
|| (warmupSeconds > 0 && rate == 0)
|| rate < 0 || numberOfMessages < 0
|| (rate == 0 && numberOfMessages == 0) || payloadBytes < TelemetryMessage.HEADER_BYTES
|| deviceCount < 1 || gatewayCount < 1 || topicCount < 1 || applicationCount < 1
|| clientsPerApplication < 1 || ioThreads < 1 || listenerThreads < 1
- || maxOutstanding < 1 || consumerTimeoutSeconds < durationSeconds + warmupSeconds
+ || maxOutstanding < 1 || consumerTimeoutSeconds < minimumRuntimeSeconds
|| clientRestartIntervalSeconds < 0 || clientRestartFraction < 0 || clientRestartFraction > 1) {
throw new IllegalArgumentException("IoT scenario counts and sizes are invalid");
}
@@ -53,6 +60,10 @@ public record IotScenario(String serviceUrl, String topicPrefix, String subscrip
}
public long warmupMessageCount() {
+ return Math.multiplyExact(warmupMessageCountPerRound(), warmupRounds);
+ }
+
+ public long warmupMessageCountPerRound() {
return warmupMessages > 0
? warmupMessages
: Math.multiplyExact((long) warmupSeconds, rate);
diff --git a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryProducer.java b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryProducer.java
index 46d17a60505c..dac9f12e3d06 100644
--- a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryProducer.java
+++ b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryProducer.java
@@ -79,6 +79,7 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
long nextSend = System.nanoTime();
long startedNanos = nextSend;
long warmupMessageCount = scenario.warmupMessageCount();
+ long warmupMessagesPerRound = scenario.warmupMessageCountPerRound();
long measurementStartedNanos = -1;
long measurementStartEpochMs = -1;
for (long sent = 0; sent < scenario.messageCount(); sent++) {
@@ -136,8 +137,24 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
if (wait > 0) {
LockSupport.parkNanos(wait);
}
+ if (!measurementMessage && warmupMessagesPerRound > 0
+ && (sent + 1) % warmupMessagesPerRound == 0) {
+ awaitOutstanding(outstanding, scenario.maxOutstanding());
+ if (failure.get() != null) {
+ throw new IllegalStateException("Telemetry warmup send failed", failure.get());
+ }
+ int round = Math.toIntExact((sent + 1) / warmupMessagesPerRound);
+ System.out.println("WARMUP_ROUND_COMPLETE round=" + round + "/" + scenario.warmupRounds()
+ + " completed=" + warmupCompleted.get()
+ + " delaySeconds=" + scenario.warmupRoundDelaySeconds());
+ if (scenario.warmupRoundDelaySeconds() > 0) {
+ TimeUnit.SECONDS.sleep(scenario.warmupRoundDelaySeconds());
+ }
+ // Do not turn time spent draining or paused into a rate-limiter catch-up burst.
+ nextSend = System.nanoTime();
+ }
}
- outstanding.acquire(scenario.maxOutstanding());
+ awaitOutstanding(outstanding, scenario.maxOutstanding());
if (failure.get() != null) {
throw new IllegalStateException("Telemetry send failed", failure.get());
}
@@ -149,6 +166,9 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
Files.writeString(output.resolve("producer-summary.json"),
"{\n \"sent\": " + completed.get()
+ ",\n \"warmupMessages\": " + warmupCompleted.get()
+ + ",\n \"warmupMessagesPerRound\": " + warmupMessagesPerRound
+ + ",\n \"warmupRounds\": " + scenario.warmupRounds()
+ + ",\n \"warmupRoundDelaySeconds\": " + scenario.warmupRoundDelaySeconds()
+ ",\n \"measurementMessages\": " + measurementCompleted.get()
+ ",\n \"devices\": " + scenario.deviceCount()
+ ",\n \"elapsedSeconds\": " + elapsedNanos / 1_000_000_000.0
@@ -174,6 +194,11 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
return 0;
}
+ private static void awaitOutstanding(Semaphore outstanding, int permits) throws InterruptedException {
+ outstanding.acquire(permits);
+ outstanding.release(permits);
+ }
+
private Producer<byte[]> createProducer(IotScenario scenario, List<PulsarClient> clients,
int gateway, int topic) throws Exception {
return clients.get(gateway).newProducer()
diff --git a/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/IotScenarioTest.java b/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/IotScenarioTest.java
index 4cb0b5f024d7..20c2d2ce625c 100644
--- a/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/IotScenarioTest.java
+++ b/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/IotScenarioTest.java
@@ -41,6 +41,24 @@ public class IotScenarioTest {
assertThat(scenario.messageCount()).isEqualTo(6_000_000);
}
+ @Test
+ public void calculatesMultipleWarmupRounds() {
+ IotScenario scenario = scenario(0, 500_000, 2, 5, 0, 5_000_000);
+
+ assertThat(scenario.warmupMessageCountPerRound()).isEqualTo(500_000);
+ assertThat(scenario.warmupMessageCount()).isEqualTo(1_000_000);
+ assertThat(scenario.messageCount()).isEqualTo(6_000_000);
+ assertThat(scenario.warmupRoundDelaySeconds()).isEqualTo(5);
+ }
+
+ @Test
+ public void defaultsMissingWarmupRoundsToOne() {
+ IotScenario scenario = scenario(0, 1_000, 0, 0, 0, 5_000);
+
+ assertThat(scenario.warmupRounds()).isEqualTo(1);
+ assertThat(scenario.warmupMessageCount()).isEqualTo(1_000);
+ }
+
@Test
public void rejectsTimeBasedWarmupWithoutRateLimit() {
assertThatThrownBy(() -> scenario(20, 0, 0, 5_000_000))
@@ -54,8 +72,14 @@ public class IotScenarioTest {
}
private static IotScenario scenario(int warmupSeconds, long warmupMessages, int rate, long numberOfMessages) {
+ return scenario(warmupSeconds, warmupMessages, 1, 0, rate, numberOfMessages);
+ }
+
+ private static IotScenario scenario(int warmupSeconds, long warmupMessages, int warmupRounds,
+ int warmupRoundDelaySeconds, int rate, long numberOfMessages) {
return new IotScenario("pulsar://localhost:6650", "persistent://public/default/iot-", "app-",
- 120, warmupSeconds, warmupMessages, rate, numberOfMessages, 64, 1_000, 10, 2, 1, 2,
+ 120, warmupSeconds, warmupMessages, warmupRounds, warmupRoundDelaySeconds,
+ rate, numberOfMessages, 64, 1_000, 10, 2, 1, 2,
2, 2, 100, true, true, 300, 0, 0);
}
}