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:53 UTC
(pulsar) 08/09: [improve][test] Bound measurement profiles by consumer delivery
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 73a29a98af7131d1a96a03f9a14b6caa8ee6b90b
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Sun Sep 20 22:19:13 2026 +0300
[improve][test] Bound measurement profiles by consumer delivery
---
tests/performance/README.md | 18 ++++---
tests/performance/iot-telemetry.md | 24 +++++----
.../launcher/JfrRecordingProcessor.java | 4 +-
.../performance/launcher/PerformanceLauncher.java | 35 +++++++++++--
.../tests/performance/launcher/JfrCutTest.java | 5 +-
.../tests/performance/tools/PerformanceTool.java | 4 ++
.../tests/performance/tools/TelemetryConsumer.java | 46 +++++++++++++---
.../tests/performance/tools/TelemetryProducer.java | 5 +-
.../tests/performance/tools/WarmupBarrier.java | 61 ++++++++++++++++++++++
.../tests/performance/tools/WarmupBarrierTest.java | 45 ++++++++++++++++
10 files changed, 212 insertions(+), 35 deletions(-)
diff --git a/tests/performance/README.md b/tests/performance/README.md
index 4fa638b543aa..cec71ab9982a 100644
--- a/tests/performance/README.md
+++ b/tests/performance/README.md
@@ -83,10 +83,13 @@ 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. `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 measurement boundaries in
-`producer-summary.json` cover only the configured measurement messages. The launcher uses the start boundary to
-remove warmup from the derived JFR while retaining everything recorded afterward.
+the final round. A round is fully drained only after every backend application has uniquely received its cumulative
+warmup message count; producer send completions alone do not release the barrier. The default is one round with no
+delay. Warmup traffic remains part of delivery and ordering validation. Producer throughput and the
+epoch-millisecond measurement boundaries in
+`producer-summary.json` cover only the configured measurement messages. Every consumer summary records its first
+and last measured-message receipt as metadata. The launcher cuts from the producer measurement start through the
+latest last receipt across all backend applications.
Use a top-level `extends` entry to inherit one file or an ordered list of files:
@@ -112,9 +115,10 @@ an inherited entry. Cycles, missing files, non-mapping roots and invalid `extend
Profiled standalone runs retain the complete JFR and also create a sibling whose name ends in
`.measurement.jfr`. The measurement recording contains events from the producer's recorded measurement start through
-the end of each process recording, excluding startup and warmup work without dropping asynchronous completion,
-consumer validation, or shutdown activity after the producer finishes sending. One-time JVM, host, recording setting and runtime configuration events
-are copied from the beginning of the complete recording so JDK Mission Control can describe the source JVM. Set
+the latest measured-message receipt across all backend applications. This excludes startup, warmup, and shutdown
+while retaining the broker and consumer work needed to deliver every measured message. One-time JVM, host, recording
+setting and runtime configuration events are copied from the beginning of the complete recording so JDK Mission
+Control can describe the source JVM. 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.
diff --git a/tests/performance/iot-telemetry.md b/tests/performance/iot-telemetry.md
index 43ec9ad05f68..90478a4a910a 100644
--- a/tests/performance/iot-telemetry.md
+++ b/tests/performance/iot-telemetry.md
@@ -65,13 +65,14 @@ 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.
-The launcher uses the start boundary to remove warmup from broker and client JFRs and retains the rest of each
-recording. Both boundaries remain available for throughput calculations and manually selecting an exact interval.
+The value applies to each of `warmupRounds`. Every round drains its asynchronous sends and waits until every backend
+application has uniquely received the cumulative warmup count 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 producer boundaries. Each
+`consumer-summary.json` records the first and last measured-message receipt as metadata. The launcher uses the
+producer start and the latest last receipt across all backend applications as the JFR measurement interval.
The base scenario also keeps incidental storage maintenance outside normal measurement windows. Its managed-ledger
entry, size and time limits allow the topic and cursor ledgers to remain open throughout ordinary runs. BookKeeper
@@ -103,10 +104,11 @@ 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 start from `producer-summary.json` to
-write a sibling `.measurement.jfr` containing the remainder of each process recording. This removes startup and
-warmup while retaining asynchronous completion, consumer validation, and shutdown activity after the producer
-finishes sending. The complete recording is retained by default. The cut recording also retains the one-time JVM, host, recording setting and runtime
+After every profiled process exits, the launcher writes a sibling `.measurement.jfr` spanning the producer's
+measurement start through the latest measured-message receipt across all backend applications. The upper boundary
+includes the full millisecond containing that receipt. This removes startup, warmup, and shutdown while retaining
+the broker and consumer work needed to deliver every measured message. The complete recording is retained by
+default. The cut recording also retains the one-time JVM, host, recording setting and runtime
configuration events needed to describe the source JVM in JDK Mission Control. 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
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
index 84a66d6aeac1..856689964674 100644
--- 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
@@ -48,13 +48,13 @@ final class JfrRecordingProcessor {
}
}
- static void process(Set<Path> recordings, Instant from,
+ 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.cutFrom(recording, from, measurementPath(recording));
+ JfrCut.cut(recording, from, to, measurementPath(recording));
}
if (!retainOriginal) {
Files.delete(recording);
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 4ece3c2af4eb..b685fa547851 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
@@ -51,6 +51,7 @@ public class PerformanceLauncher implements Callable<Integer> {
private static final String CONFIG_ENV = "PULSAR_PERFORMANCE_CONFIG";
private static final String TOOLS_MOUNT = "/opt/pulsar-performance-tools";
private static final String CONFIG_MOUNT = "/performance-config/resolved-config.yaml";
+ private static final String COORDINATION_MOUNT = "/performance-coordination";
@Option(names = "--config", required = true)
Path config;
@@ -92,6 +93,16 @@ public class PerformanceLauncher implements Callable<Integer> {
: Path.of(loader.select(resolved, "output.directory").textValue());
runOutput = runOutput.toAbsolutePath().normalize();
Files.createDirectories(runOutput);
+ Path coordinationDirectory = runOutput.resolve("coordination");
+ Files.createDirectories(coordinationDirectory);
+ try (var existingMarkers = Files.list(coordinationDirectory)) {
+ for (Path marker : existingMarkers
+ .filter(path -> path.getFileName().toString().startsWith("warmup-round-"))
+ .filter(path -> path.getFileName().toString().endsWith(".complete"))
+ .toList()) {
+ Files.deleteIfExists(marker);
+ }
+ }
Set<Path> recordingsBeforeRun = JfrRecordingProcessor.findOriginalRecordings(runOutput);
Path brokerProfileDirectory = runOutput.resolve("broker-profile");
if (brokerProfileOptions != null) {
@@ -134,7 +145,8 @@ public class PerformanceLauncher implements Callable<Integer> {
for (int application = 0; application < applications; application++) {
Path appOutput = runOutput.resolve("consumer-" + application);
Files.createDirectories(appOutput);
- consumers.add(workloadContainer(cluster, resolvedToolsDirectory, resolvedConfig, appOutput,
+ consumers.add(workloadContainer(cluster, resolvedToolsDirectory, resolvedConfig,
+ coordinationDirectory, appOutput,
consumerProfileOptions, "iot-consume", "--application-index", Integer.toString(application))
.waitingFor(Wait.forLogMessage(".*READY application=.*", 1)
.withStartupTimeout(Duration.ofMinutes(5))));
@@ -143,7 +155,8 @@ public class PerformanceLauncher implements Callable<Integer> {
Path producerOutput = runOutput.resolve("producer");
Files.createDirectories(producerOutput);
- producer = workloadContainer(cluster, resolvedToolsDirectory, resolvedConfig, producerOutput,
+ producer = workloadContainer(cluster, resolvedToolsDirectory, resolvedConfig,
+ coordinationDirectory, producerOutput,
producerProfileOptions, "iot-produce");
producer.start();
int timeout = workload.path("consumerTimeoutSeconds").intValue() + 60;
@@ -176,18 +189,29 @@ public class PerformanceLauncher implements Callable<Integer> {
if (profilingEnabled) {
JsonNode summary = loader.mapper().readTree(runOutput.resolve("producer/producer-summary.json").toFile());
Instant measurementStart = Instant.ofEpochMilli(requiredLong(summary, "measurementStartEpochMs"));
+ long lastConsumerReceiptEpochMs = Long.MIN_VALUE;
+ for (int application = 0; application < applications; application++) {
+ JsonNode consumerSummary = loader.mapper().readTree(
+ runOutput.resolve("consumer-" + application + "/consumer-summary.json").toFile());
+ lastConsumerReceiptEpochMs = Math.max(lastConsumerReceiptEpochMs,
+ requiredLong(consumerSummary, "lastMeasurementMessageReceivedEpochMs"));
+ }
+ // Consumer timestamps have millisecond precision. Use the following millisecond as the exclusive bound
+ // so that events from the millisecond containing the final receipt are retained.
+ Instant measurementEnd = Instant.ofEpochMilli(lastConsumerReceiptEpochMs).plusMillis(1);
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,
+ JfrRecordingProcessor.process(recordings, measurementStart, measurementEnd,
retainOriginalRecording, createMeasurementRecording);
}
return 0;
}
private GenericContainer<?> workloadContainer(PulsarCluster cluster, Path tools, Path configFile,
+ Path coordinationDirectory,
Path outputDirectory, String profileOptions,
String command, String... extraArguments) {
List<String> arguments = new ArrayList<>();
@@ -197,6 +221,8 @@ public class PerformanceLauncher implements Callable<Integer> {
arguments.add(CONFIG_MOUNT);
arguments.add("--output");
arguments.add("/performance-output");
+ arguments.add("--coordination-directory");
+ arguments.add(COORDINATION_MOUNT);
arguments.addAll(List.of(extraArguments));
String javaOptions = "-Xms128m -Xmx512m -XX:MaxDirectMemorySize=256m";
if (profileOptions != null) {
@@ -211,6 +237,7 @@ public class PerformanceLauncher implements Callable<Integer> {
.withNetwork(cluster.getNetwork())
.withFileSystemBind(tools.toString(), TOOLS_MOUNT, BindMode.READ_ONLY)
.withFileSystemBind(configFile.toString(), CONFIG_MOUNT, BindMode.READ_ONLY)
+ .withFileSystemBind(coordinationDirectory.toString(), COORDINATION_MOUNT, BindMode.READ_WRITE)
.withFileSystemBind(outputDirectory.toString(), "/performance-output", BindMode.READ_WRITE)
.withEnv("JAVA_TOOL_OPTIONS", javaOptions)
.withCommand(arguments.toArray(String[]::new));
@@ -242,7 +269,7 @@ public class PerformanceLauncher implements Callable<Integer> {
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);
+ throw new IllegalArgumentException("Missing numeric performance summary field " + field);
}
return value.longValue();
}
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
index ac867a17820a..70b275d3f4e1 100644
--- 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
@@ -72,12 +72,13 @@ public class JfrCutTest {
Instant[] interval = createRecording(input);
Path measurement = JfrRecordingProcessor.measurementPath(input);
- JfrRecordingProcessor.process(Set.of(input), interval[0], retainOriginal, createMeasurement);
+ JfrRecordingProcessor.process(Set.of(input), interval[0], interval[1],
+ retainOriginal, createMeasurement);
assertEquals(Files.exists(input), retainOriginal);
assertEquals(Files.exists(measurement), createMeasurement);
if (createMeasurement) {
- assertEquals(markers(measurement), List.of("measurement", "after"));
+ assertEquals(markers(measurement), List.of("measurement"));
}
} finally {
deleteDirectory(directory);
diff --git a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/PerformanceTool.java b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/PerformanceTool.java
index 0bfc10742dae..4182c5861de1 100644
--- a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/PerformanceTool.java
+++ b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/PerformanceTool.java
@@ -55,6 +55,10 @@ public class PerformanceTool implements Callable<Integer> {
@Option(names = "--output", required = true)
Path output;
+ @Option(names = "--coordination-directory", required = true,
+ description = "Shared directory for workload phase barriers")
+ Path coordinationDirectory;
+
IotScenario scenario() throws Exception {
return readScenario(config, configPath);
}
diff --git a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryConsumer.java b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryConsumer.java
index c5f84fdbb849..e4d8ede9caf0 100644
--- a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryConsumer.java
+++ b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryConsumer.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.tests.performance.tools;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.time.Duration;
@@ -26,6 +27,7 @@ import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.Consumer;
@@ -53,19 +55,20 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
HdrLatencyRecorder receiveLatency = new HdrLatencyRecorder();
AtomicLong firstMeasurementReceiptEpochMs = new AtomicLong();
AtomicLong lastMeasurementReceiptEpochMs = new AtomicLong();
+ AtomicInteger nextWarmupRound = new AtomicInteger(1);
Thread restarter = null;
PulsarClientSharedResources sharedResources = SharedClientResources.create(scenario);
try {
for (int pod = 0; pod < scenario.clientsPerApplication(); pod++) {
pods.add(createPod(scenario, sharedResources, tracker, receiveLatency,
- firstMeasurementReceiptEpochMs, lastMeasurementReceiptEpochMs, pod));
+ firstMeasurementReceiptEpochMs, lastMeasurementReceiptEpochMs, nextWarmupRound, pod));
}
System.out.println("READY application=" + applicationIndex + " clients=" + pods.size());
if (scenario.clientRestartIntervalSeconds() > 0 && scenario.clientRestartFraction() > 0) {
restarter = new Thread(() -> restartClients(scenario, sharedResources, tracker, pods, stopping,
restarterFailure, receiveLatency, firstMeasurementReceiptEpochMs,
- lastMeasurementReceiptEpochMs),
+ lastMeasurementReceiptEpochMs, nextWarmupRound),
"iot-client-restarter");
restarter.start();
}
@@ -95,7 +98,11 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
+ " \"uniqueMessages\": " + summary.uniqueMessages() + ",\n"
+ " \"duplicates\": " + summary.duplicates() + ",\n"
+ " \"orderingViolations\": " + summary.orderingViolations() + ",\n"
- + " \"invalidMessages\": " + summary.invalidMessages() + "\n}\n");
+ + " \"invalidMessages\": " + summary.invalidMessages() + ",\n"
+ + " \"firstMeasurementMessageReceivedEpochMs\": "
+ + firstMeasurementReceiptEpochMs.get() + ",\n"
+ + " \"lastMeasurementMessageReceivedEpochMs\": "
+ + lastMeasurementReceiptEpochMs.get() + "\n}\n");
return summary.valid() && summary.uniqueMessages() == scenario.messageCount() ? 0 : 1;
} finally {
stopping.set(true);
@@ -111,7 +118,8 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
private ClientAndConsumer createPod(IotScenario scenario, PulsarClientSharedResources sharedResources,
DeviceSequenceTracker tracker, HdrLatencyRecorder receiveLatency,
AtomicLong firstMeasurementReceiptEpochMs,
- AtomicLong lastMeasurementReceiptEpochMs, int podIndex) throws Exception {
+ AtomicLong lastMeasurementReceiptEpochMs, AtomicInteger nextWarmupRound,
+ int podIndex) throws Exception {
PulsarClient client = PulsarClient.builder()
.serviceUrl(scenario.serviceUrl())
.sharedResources(sharedResources)
@@ -124,6 +132,7 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
.subscriptionType(SubscriptionType.Key_Shared)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.messageListener((currentConsumer, message) -> {
+ long receivedEpochMs = System.currentTimeMillis();
try {
TelemetryMessage.Decoded decoded = TelemetryMessage.decode(message.getData());
byte[] key = message.getKeyBytes();
@@ -132,13 +141,14 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
throw new IllegalArgumentException("Telemetry key does not match payload device ID");
}
if (decoded.measurement()) {
- long receivedEpochMs = System.currentTimeMillis();
- firstMeasurementReceiptEpochMs.compareAndSet(0, receivedEpochMs);
+ firstMeasurementReceiptEpochMs.accumulateAndGet(receivedEpochMs,
+ (current, received) -> current == 0 ? received : Math.min(current, received));
lastMeasurementReceiptEpochMs.accumulateAndGet(receivedEpochMs, Math::max);
receiveLatency.recordMillis(receivedEpochMs - message.getPublishTime());
}
tracker.received(decoded.deviceId(), decoded.sequence(), message.getMessageId(),
decoded.sentNanos(), message.getTopicName(), Thread.currentThread().getName());
+ markCompletedWarmupRounds(scenario, tracker, nextWarmupRound);
currentConsumer.acknowledgeAsync(message);
} catch (RuntimeException error) {
tracker.invalidMessage();
@@ -157,7 +167,7 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
DeviceSequenceTracker tracker, List<ClientAndConsumer> pods,
AtomicBoolean stopping, AtomicReference<Throwable> failure,
HdrLatencyRecorder receiveLatency, AtomicLong firstMeasurementReceiptEpochMs,
- AtomicLong lastMeasurementReceiptEpochMs) {
+ AtomicLong lastMeasurementReceiptEpochMs, AtomicInteger nextWarmupRound) {
int restartCount = Math.max(1,
(int) Math.ceil(scenario.clientsPerApplication() * scenario.clientRestartFraction()));
while (!stopping.get()) {
@@ -169,7 +179,8 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
ClientAndConsumer previous = pods.get(index);
previous.close();
pods.set(index, createPod(scenario, sharedResources, tracker, receiveLatency,
- firstMeasurementReceiptEpochMs, lastMeasurementReceiptEpochMs, index));
+ firstMeasurementReceiptEpochMs, lastMeasurementReceiptEpochMs, nextWarmupRound,
+ index));
}
}
} catch (InterruptedException interrupted) {
@@ -182,6 +193,25 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
}
}
+ private void markCompletedWarmupRounds(IotScenario scenario, DeviceSequenceTracker tracker,
+ AtomicInteger nextWarmupRound) {
+ long messagesPerRound = scenario.warmupMessageCountPerRound();
+ if (messagesPerRound == 0) {
+ return;
+ }
+ int round;
+ while ((round = nextWarmupRound.get()) <= scenario.warmupRounds()
+ && tracker.uniqueMessages() >= Math.multiplyExact(messagesPerRound, round)) {
+ if (nextWarmupRound.compareAndSet(round, round + 1)) {
+ try {
+ WarmupBarrier.markApplicationComplete(coordinationDirectory, round, applicationIndex);
+ } catch (IOException error) {
+ throw new IllegalStateException("Cannot mark warmup round complete", error);
+ }
+ }
+ }
+ }
+
private record ClientAndConsumer(PulsarClient client, Consumer<byte[]> consumer) implements AutoCloseable {
@Override
public void close() throws Exception {
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 85e4943e7f1b..a81775326357 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
@@ -148,8 +148,11 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
throw new IllegalStateException("Telemetry warmup send failed", failure.get());
}
int round = Math.toIntExact((sent + 1) / warmupMessagesPerRound);
+ WarmupBarrier.awaitApplications(coordinationDirectory, round, scenario.applicationCount(),
+ scenario.consumerTimeoutSeconds());
System.out.println("WARMUP_ROUND_COMPLETE round=" + round + "/" + scenario.warmupRounds()
- + " completed=" + warmupCompleted.get()
+ + " produced=" + warmupCompleted.get()
+ + " applicationsReceived=" + scenario.applicationCount()
+ " delaySeconds=" + scenario.warmupRoundDelaySeconds());
if (scenario.warmupRoundDelaySeconds() > 0) {
TimeUnit.SECONDS.sleep(scenario.warmupRoundDelaySeconds());
diff --git a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/WarmupBarrier.java b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/WarmupBarrier.java
new file mode 100644
index 000000000000..29b58c89c215
--- /dev/null
+++ b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/WarmupBarrier.java
@@ -0,0 +1,61 @@
+/*
+ * 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.tools;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+
+/** Coordinates warmup delivery without adding per-message work after warmup has completed. */
+final class WarmupBarrier {
+ private static final Duration POLL_INTERVAL = Duration.ofMillis(20);
+
+ private WarmupBarrier() {
+ }
+
+ static void markApplicationComplete(Path directory, int round, int application) throws IOException {
+ Files.createDirectories(directory);
+ Files.writeString(marker(directory, round, application), "complete\n");
+ }
+
+ static void awaitApplications(Path directory, int round, int applications, int timeoutSeconds)
+ throws Exception {
+ long deadline = System.nanoTime() + Duration.ofSeconds(timeoutSeconds).toNanos();
+ while (System.nanoTime() < deadline) {
+ boolean complete = true;
+ for (int application = 0; application < applications; application++) {
+ if (!Files.isRegularFile(marker(directory, round, application))) {
+ complete = false;
+ break;
+ }
+ }
+ if (complete) {
+ return;
+ }
+ Thread.sleep(POLL_INTERVAL.toMillis());
+ }
+ throw new IllegalStateException("Timed out waiting for all " + applications
+ + " applications to receive warmup round " + round);
+ }
+
+ private static Path marker(Path directory, int round, int application) {
+ return directory.resolve("warmup-round-" + round + "-application-" + application + ".complete");
+ }
+}
diff --git a/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/WarmupBarrierTest.java b/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/WarmupBarrierTest.java
new file mode 100644
index 000000000000..45e051fdde73
--- /dev/null
+++ b/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/WarmupBarrierTest.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.performance.tools;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import org.testng.annotations.Test;
+
+public class WarmupBarrierTest {
+ @Test
+ public void waitsForEveryApplicationInTheRound() throws Exception {
+ Path directory = Files.createTempDirectory("warmup-barrier-test");
+ try {
+ WarmupBarrier.markApplicationComplete(directory, 2, 0);
+ WarmupBarrier.markApplicationComplete(directory, 2, 1);
+
+ assertThatCode(() -> WarmupBarrier.awaitApplications(directory, 2, 2, 1))
+ .doesNotThrowAnyException();
+ } finally {
+ try (var paths = Files.walk(directory)) {
+ for (Path path : paths.sorted(Comparator.reverseOrder()).toList()) {
+ Files.deleteIfExists(path);
+ }
+ }
+ }
+ }
+}