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/10/19 21:18:21 UTC
(pulsar) 02/16: [improve][test] Add Key_Shared replay queue and stuck consumer scenarios to the performance tests (#26975)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8e962ba419a0e6548e915788c706121705fcf6d0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Oct 19 11:59:45 2026 +0300
[improve][test] Add Key_Shared replay queue and stuck consumer scenarios to the performance tests (#26975)
(cherry picked from commit 13ed8f06846ef6b27f3e702430a10566bbeb2e76)
---
tests/performance/AGENTS.md | 4 +-
tests/performance/README.md | 4 +-
tests/performance/docs/run-reports.md | 9 ++-
tests/performance/docs/running-scenarios.md | 7 +-
.../performance/launcher/ProgressMonitor.java | 6 +-
.../performance/launcher/TopicStatsSampler.java | 11 ++--
.../performance/launcher/ProgressMonitorTest.java | 12 ++--
.../tests/performance/report/ComparisonCharts.java | 2 +-
.../pulsar/tests/performance/report/RunReport.java | 76 +++++++++++++++++++---
.../tests/performance/report/RunReportTest.java | 24 ++++++-
tests/performance/scenarios/docs/iot-telemetry.md | 23 +++++++
.../scenarios/iot-telemetry-key-shared-replay.yaml | 56 ++++++++++++++++
.../iot-telemetry-key-shared-stuck-consumer.yaml | 37 +++++++++++
.../tests/performance/tools/IotScenario.java | 67 +++++++++++++++++--
.../tests/performance/tools/TelemetryConsumer.java | 72 ++++++++++++++++----
.../tests/performance/tools/TelemetryMessage.java | 22 +++++++
.../tests/performance/tools/TelemetryProducer.java | 6 +-
.../tests/performance/tools/IotScenarioTest.java | 47 +++++++++++++
.../performance/tools/TelemetryMessageTest.java | 43 ++++++++++++
19 files changed, 474 insertions(+), 54 deletions(-)
diff --git a/tests/performance/AGENTS.md b/tests/performance/AGENTS.md
index 441336715d4d..d26cf85f4b7a 100644
--- a/tests/performance/AGENTS.md
+++ b/tests/performance/AGENTS.md
@@ -365,7 +365,7 @@ commit segment can be absent). Use the exact recording names linked from the pro
| `gateways/container.log.txt`, `applications/container.log.txt` | The workload containers' logs | Read |
| `gateways/gateways-latency.hgrm`, `applications/<application>/application-latency.hgrm` | The publish and end-to-end latency percentile distributions, in milliseconds, as text | Read |
| `gateways/gateways-latency.hdr`, `applications/<application>/application-latency.hdr` | The latency interval logs | `renderHdrHistograms`, HistogramLogAnalyzer |
-| `topic-stats.csv` | The broker's topic stats, sampled once per second, or every topics / 100 seconds with more than 100 topics: backlog and message counters per subscription | DuckDB, or any CSV reader |
+| `topic-stats.csv` | The broker's topic stats, sampled once per second, or every topics / 100 seconds with more than 100 topics: backlog, message counters and entries in replay per subscription | DuckDB, or any CSV reader |
| `join-preparation.csv` | When the launcher prepared the late applications' join, each skip, reset and unload, and how long it took, see [Offloading](docs/offloading.md) | DuckDB, or any CSV reader |
| `offload-status.csv` | In a run that offloads to tiered storage, each topic's ledgers, offloaded ledgers and offloaded bytes when the late applications joined and at the run's end | DuckDB, or any CSV reader |
| `offload-storage.log.txt` | In a run that offloads to tiered storage, the storage's log, with every S3 request when `offload.logRequests` is set | `rg` |
@@ -374,7 +374,7 @@ commit segment can be absent). Use the exact recording names linked from the pro
| `host-io.csv` | The Docker engine host's CPU busy and I/O-wait share and each disk's read/write MB/s and busy share, sampled once per second; on macOS, the Docker VM's, read through the perf stat sidecar. CPUs near 100 % busy delay every thread; a disk near 100 % busy limits the bookies | DuckDB, or any CSV reader |
| `container-summary.json` | **Start here for CPU use per component.** The measurement's averages: the host's CPU busy and I/O-wait share and disk MB/s, and for each container and all of them the CPUs, CPU seconds per million messages, voluntary and involuntary context switches, CPU migrations and page faults per second, the clock rate, IPC, and last-level cache, L1 data cache and branch misses per thousand instructions | `jq` |
| `container-stats.csv`, `perf-stat.csv` | The per-second samples behind the summary: each container's CPUs and context switches, and its `perf stat` counts (task-clock, context switches, migrations, page faults, cycles, instructions, cache references and misses, L1 data cache load misses, branch misses) | DuckDB, or any CSV reader |
-| `throughput.svg`, `backlog.svg`, `latency-percentiles.svg`, `latency-timeline.svg`, `host-temperature.svg`, `host-frequency.svg` | The report's charts, each with a PNG beside it | View the PNG |
+| `throughput.svg`, `backlog.svg`, `replay.svg`, `latency-percentiles.svg`, `latency-timeline.svg`, `host-temperature.svg`, `host-frequency.svg` | The report's charts, each with a PNG beside it | View the PNG |
| `metrics.json` | What querying the run's metrics needs: its label selector, time range, jobs and instances, and the URLs and credentials of VictoriaMetrics and Grafana, see [metrics.json](docs/metrics.md#metricsjson) | `jq`, then VictoriaMetrics' Prometheus API |
| `grafana-panels/*.png` | Panels of Grafana's dashboards over the run | View |
| `network-events.jsonl`, `network-summary.json` | With a `network` section: every impairment's application, removal and queue counters as they happened, and each impaired container's TCP segments sent and retransmitted, see [Network conditions](docs/network-conditions.md#files) | `jq` |
diff --git a/tests/performance/README.md b/tests/performance/README.md
index 75d2802bf7f9..f9b9bd97640c 100644
--- a/tests/performance/README.md
+++ b/tests/performance/README.md
@@ -86,7 +86,7 @@ the run report are written for the IoT domain.
| Domain | Simulation |
|---|---|
-| Device | A device ID, which is the message key. `devices.count` sets the number of devices |
+| Device | A device ID, which is the message key. `devices.count` sets the number of devices, and `devices.keyType` whether the key is the ID as bytes, the default, or `device-<ID>` as a string |
| Telemetry message | A message with the device ID, the device's sequence number and the send time, `payload.size` bytes long |
| Gateway | A Pulsar client in the gateways' container, with a producer named `iot-gateway-<gateway>-topic-<topic>` for each topic. `gateways.count` sets the number of gateways; their clients share I/O threads and memory, as the clients of one process can since PIP-234 |
| Topics | `topics.count` topics; a device's messages always go to the same topic, the device ID modulo `topics.count` |
@@ -228,7 +228,7 @@ Run directory: .../build/performance/2026-09-26/master/iot-telemetry/09-26-12-00
12:00:49 Starting the gateways: 20,000 warmup and 120,000 measured message(s) at 1,000 msg/s from 100 gateway(s) to 30 topic(s)
...
[01:12 measurement 40 s] Produced: 60,548 msg of 140,000 (43%) --- 1,000.9 msg/s --- 0.5 Mbit/s --- pending: 1 --- Latency: mean: 2.038 ms - med: 1.392 - 95pct: 5.463 - 99pct: 11.647 - 99.9pct: 22.079 - 99.99pct: 28.031 - Max: 28.063
-[01:12 measurement 40 s] Received: 1,215,940 msg of 2,800,000 (43%) --- 20,016.4 msg/s --- 10.2 Mbit/s --- backlog: 1,161 msg (max per application: 66) --- Latency: mean: 5.361 ms - med: 2.000 - 95pct: 16.007 - 99pct: 40.031 - 99.9pct: 70.015 - 99.99pct: 81.023 - Max: 94.015
+[01:12 measurement 40 s] Received: 1,215,940 msg of 2,800,000 (43%) --- 20,016.4 msg/s --- 10.2 Mbit/s --- backlog: 1,161 entries (max per application: 66) --- Latency: mean: 5.361 ms - med: 2.000 - 95pct: 16.007 - 99pct: 40.031 - 99.9pct: 70.015 - 99.99pct: 81.023 - Max: 94.015
...
12:03:26 The gateways have finished; waiting for the applications to receive every message
12:03:34 Every application has received every message; verifying the device sequences
diff --git a/tests/performance/docs/run-reports.md b/tests/performance/docs/run-reports.md
index 508bc726320f..aa27e941e55d 100644
--- a/tests/performance/docs/run-reports.md
+++ b/tests/performance/docs/run-reports.md
@@ -104,7 +104,7 @@ didn't receive every message, and the launcher fails a run when an application's
├── <scenario>.yaml, resolved-config.yaml
├── run-info.json, run-id.txt
├── throughput.svg, backlog.svg, latency-percentiles.svg, latency-timeline.svg,
-│ host-temperature.svg, host-frequency.svg the charts, each also as PNG
+│ replay.svg, host-temperature.svg, host-frequency.svg the charts, each also as PNG
├── topic-stats.csv, host-stats.csv the sampled topic stats and host CPU
├── host-io.csv the sampled Docker engine host CPU utilization and disk throughput
├── container-stats.csv, perf-stat.csv each container's CPU use, context switches and CPU counters
@@ -169,7 +169,9 @@ report has these sections:
Correctness section, and their end-to-end latency, which includes the replication, with the remote cluster's files.
- **Backlog and rates**: each subscription's backlog and the per-second rates, sampled from the broker's topic
stats once per second, or every topics / 100 seconds with more than 100 topics, while the gateways publish and the applications receive. A sampled maximum is not the exact peak
- between samples.
+ between samples. When a subscription had entries in its replay queue, such as Key_Shared messages held back for a
+ consumer without permits, the section also shows the replay queues. The broker counts the backlog and the replay
+ queue in entries, not messages: a batch counts once, however many messages it has.
- **Host**: the CPU temperature, frequency and thermal throttling at the start and during the measurement, with
their charts in a collapsed section; a chart whose values the host doesn't provide is left out. The report says so
in bold when the CPU throttled during the measurement, and that throttling is unknown when the host has no thermal
@@ -201,10 +203,11 @@ beside it, rendered with [commonmark-java](https://github.com/commonmark/commonm
| `launcher.log` | The launcher's log: Testcontainers' log and the Pulsar containers' logs, which stay off the console. It is written during the run, and a successful run deletes it at the end, since the containers' logs make it large, unless `--keep-launcher-log` or `-Pperformance.keepLauncherLog` keeps it; a failed run keeps it, and so does a run whose applications received duplicates, ordering violations or invalid messages |
| `throughput.svg`, `.png` | Messages published and dispatched per second over the run, warmup included and the gateways' finish marked. A cool-down wait of 10 s or more before the measurement is cut out of the time axis |
| `backlog.svg`, `.png` | Each subscription's backlog over the run, on the same time axis |
+| `replay.svg`, `.png` | The entries in each subscription's replay queue over the run, on the same time axis, when a subscription had any |
| `latency-percentiles.svg`, `.png` | Latency by percentile, as HistogramLogAnalyzer plots it: publish and each application's end to end, on an axis that spreads the tail (90 %, 99 %, 99.9 %, …) |
| `latency-timeline.svg`, `.png` | The maximum latency of each logged interval over the run, publish and per application |
| `host-temperature.svg`, `.png`, `host-frequency.svg`, `.png` | The CPU package and hottest core temperature, and the mean and lowest core frequency, over the run |
-| `topic-stats.csv` | The broker's topic stats sampled once per second, or every topics / 100 seconds with more than 100 topics: backlog and message counters per subscription |
+| `topic-stats.csv` | The broker's topic stats sampled once per second, or every topics / 100 seconds with more than 100 topics: backlog, message counters and entries in replay per subscription |
| `offload-storage.log.txt` | In a run that offloads to tiered storage, the storage's log, with every request when `offload.logRequests` is set, see [Offloading](offloading.md) |
| `seeks.csv` | With an `offload.seeks` section, each seek to a random publish time: its topic, publish time, how long the reset and the read of the message took, and the position, see [Random seeks](offloading.md#random-seeks) |
| `join-preparation.csv` | When the launcher prepared the late applications' join, each skip, reset and unload, and how long it took, see [Offloading](offloading.md) |
diff --git a/tests/performance/docs/running-scenarios.md b/tests/performance/docs/running-scenarios.md
index d56a505ba1c9..6c0d377e392c 100644
--- a/tests/performance/docs/running-scenarios.md
+++ b/tests/performance/docs/running-scenarios.md
@@ -156,15 +156,16 @@ backlog and end-to-end latency. The prefix has the time since the gateways start
```
[01:21 measurement 47 s] Produced: 67,816 msg of 140,000 (48%) --- 1,020.0 msg/s --- 0.5 Mbit/s --- pending: 3 --- Latency: mean: 65.769 ms - med: 6.271 - 95pct: 342.783 - 99pct: 504.063 - 99.9pct: 744.447 - 99.99pct: 802.815 - Max: 814.591
-[01:21 measurement 47 s] Received: 1,354,097 msg of 2,800,000 (48%) --- 20,048.9 msg/s --- 10.3 Mbit/s --- backlog: 1,081 msg (max per application: 65) --- Latency: mean: 80.588 ms - med: 13.007 - 95pct: 383.231 - 99pct: 550.399 - 99.9pct: 776.191 - 99.99pct: 874.495 - Max: 921.087
+[01:21 measurement 47 s] Received: 1,354,097 msg of 2,800,000 (48%) --- 20,048.9 msg/s --- 10.3 Mbit/s --- backlog: 1,081 entries (max per application: 65) --- Latency: mean: 80.588 ms - med: 13.007 - 95pct: 383.231 - 99pct: 550.399 - 99.9pct: 776.191 - 99.99pct: 874.495 - Max: 921.087
```
- The throughput and the latencies are those of the interval since the previous lines. The received messages and
the throughput are summed over every application, and the latencies merged over every application; they include
the warmup, which the latency logs and the run report leave out. The bit rates count the payload only.
- The backlog is the sum of every subscription's backlog, from the latest topic stats sample, and the largest
- application's, summed over its topics. It includes the messages that the applications have received but whose
- acknowledgments the broker hasn't processed yet, so it isn't zero while messages flow. When the latest sample is
+ application's, summed over its topics, in entries, as the broker counts it: a batch counts once. It includes the
+ messages that the applications have received but whose acknowledgments the broker hasn't processed yet, so it
+ isn't zero while messages flow. When the latest sample is
5 seconds old or older, because the broker answers the stats requests slowly, the line says how old it is. The lines
also show duplicates and ordering violations as soon as an application has any.
- The end-to-end latency is measured from the broker's publish time, which has millisecond resolution.
diff --git a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitor.java b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitor.java
index 2721c3bc1e3d..0d07c1198796 100644
--- a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitor.java
+++ b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitor.java
@@ -222,7 +222,7 @@ final class ProgressMonitor implements AutoCloseable {
(received - lastReceived) / seconds, bitRate((received - lastReceived) / seconds)));
TopicStatsSampler.Backlog currentBacklog = backlog.get();
if (currentBacklog != null) {
- receivedLine.append(String.format(Locale.ROOT, " --- backlog: %,d msg (max per application: %,d%s)",
+ receivedLine.append(String.format(Locale.ROOT, " --- backlog: %,d entries (max per application: %,d%s)",
currentBacklog.total(), currentBacklog.maxSubscription(), age(currentBacklog)));
}
if (finishedApplications > 0 && finishedApplications < applications) {
@@ -272,13 +272,13 @@ final class ProgressMonitor implements AutoCloseable {
of(receivedTarget, received), (received - lastRemoteReceived) / seconds));
TopicStatsSampler.Backlog remote = remoteBacklog.get();
if (remote != null) {
- remoteLine.append(String.format(Locale.ROOT, " --- backlog: %,d msg (max per application: %,d%s)",
+ remoteLine.append(String.format(Locale.ROOT, " --- backlog: %,d entries (max per application: %,d%s)",
remote.total(), remote.maxSubscription(), age(remote)));
}
TopicStatsSampler.Backlog currentBacklog = backlog.get();
if (currentBacklog != null && currentBacklog.replication() >= 0) {
String age = age(currentBacklog);
- remoteLine.append(String.format(Locale.ROOT, " --- replication backlog: %,d msg%s",
+ remoteLine.append(String.format(Locale.ROOT, " --- replication backlog: %,d entries%s",
currentBacklog.replication(), age.isEmpty() ? "" : " (" + age.substring(2) + ")"));
}
if (finishedApplications > 0 && finishedApplications < remoteApplications) {
diff --git a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/TopicStatsSampler.java b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/TopicStatsSampler.java
index 833dab15d93a..08518e0e36be 100644
--- a/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/TopicStatsSampler.java
+++ b/tests/performance/launcher/src/main/java/org/apache/pulsar/tests/performance/launcher/TopicStatsSampler.java
@@ -45,10 +45,10 @@ import org.apache.pulsar.tests.performance.report.RunReport;
/**
* Samples the workload topics' stats once per second, or less often with many topics, into {@code topic-stats.csv}:
- * each subscription's backlog, the
- * topic's published-message counter and the subscription's dispatched-message counter. The report derives
- * per-second publish and dispatch rates from the counter deltas, because the broker's own rates refresh only once
- * per stats interval, and the sampled maximum backlog, which is not the exact peak between samples.
+ * each subscription's backlog, the topic's published-message counter, the subscription's dispatched-message counter
+ * and the entries in its replay queue. The report derives per-second publish and dispatch rates from the counter
+ * deltas, because the broker's own rates refresh only once per stats interval, and the sampled maximum backlog, which
+ * is not the exact peak between samples.
*
* <p>One sample is one REST call per topic, all topics in parallel, that reads the broker's in-memory counters and its
* estimated backlog (the default, not the precise backlog that scans the ledger), without the stats of the topic's
@@ -212,7 +212,8 @@ final class TopicStatsSampler implements AutoCloseable {
rows.append(',').append(topic).append(',').append(subscription.getKey()).append(',')
.append(subscription.getValue().getMsgBacklog()).append(',')
.append(stats.getMsgInCounter()).append(',')
- .append(subscription.getValue().getMsgOutCounter()).append('\n');
+ .append(subscription.getValue().getMsgOutCounter()).append(',')
+ .append(subscription.getValue().getMsgInReplay()).append('\n');
}
if (replicationWriter != null) {
for (Map.Entry<String, ? extends ReplicatorStats> replicator : stats.getReplication().entrySet()) {
diff --git a/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitorTest.java b/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitorTest.java
index 7605897237f1..5cbb40e58672 100644
--- a/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitorTest.java
+++ b/tests/performance/launcher/src/test/java/org/apache/pulsar/tests/performance/launcher/ProgressMonitorTest.java
@@ -59,7 +59,7 @@ public class ProgressMonitorTest {
assertThat(lines[0]).contains("warmup round 1/1", "Produced: 1,000 msg of 4,000 (25%)", "pending: 7",
"Latency: mean: 2.000 ms");
assertThat(lines[1]).contains("Received: 1,000 msg of 8,000 (12%)",
- "backlog: 1,500 msg (max per application: 1,000)", "finished applications: 1/2", "med: 10.0",
+ "backlog: 1,500 entries (max per application: 1,000)", "finished applications: 1/2", "med: 10.0",
"Max: 20.0");
}
@@ -85,8 +85,8 @@ public class ProgressMonitorTest {
assertThat(lines).hasSize(3);
assertThat(lines[1]).contains("Received: 1,000 msg of 8,000 (12%)", "Max: 10.0").doesNotContain("duplicates");
assertThat(lines[2]).contains("Replicated, received in the remote cluster: 600 msg of 16,000",
- "backlog: 8,265 msg (max per application: 2,100)",
- "replication backlog: 2,345 msg", "finished applications: 1/4", "duplicates: 3", "Max: 90.0");
+ "backlog: 8,265 entries (max per application: 2,100)",
+ "replication backlog: 2,345 entries", "finished applications: 1/4", "duplicates: 3", "Max: 90.0");
}
@Test
@@ -101,8 +101,8 @@ public class ProgressMonitorTest {
monitor.report();
}
assertThat(output.toString(StandardCharsets.UTF_8))
- .contains("backlog: 0 msg (max per application: 0, sampled 30 s ago)",
- "replication backlog: 400 msg (sampled 20 s ago)");
+ .contains("backlog: 0 entries (max per application: 0, sampled 30 s ago)",
+ "replication backlog: 400 entries (sampled 20 s ago)");
}
@Test
@@ -115,7 +115,7 @@ public class ProgressMonitorTest {
monitor.report();
}
assertThat(output.toString(StandardCharsets.UTF_8))
- .contains("backlog: 1,241 msg (max per application: 73, sampled 30 s ago)");
+ .contains("backlog: 1,241 entries (max per application: 73, sampled 30 s ago)");
}
@Test
diff --git a/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/ComparisonCharts.java b/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/ComparisonCharts.java
index 7889f0def62f..5d0301c5d72f 100644
--- a/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/ComparisonCharts.java
+++ b/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/ComparisonCharts.java
@@ -207,7 +207,7 @@ public final class ComparisonCharts implements Callable<Integer> {
TimeSeriesRenderer.axisMaximum(max(throughput, time) * 1.05), false, 0, throughput, markers),
a, b, footer);
write(written, output, BACKLOG + labels, new Chart("Backlog",
- "Messages in the backlog, sampled from the topic stats", time,
+ "Entries in the backlog, sampled from the topic stats", time,
TimeSeriesRenderer.axisMaximum(max(backlog, time) * 1.05), false, 0, backlog, markers),
a, b, footer);
write(written, output, LATENCY + labels, new Chart("Latency by percentile", "Latency (ms)", percentiles,
diff --git a/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/RunReport.java b/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/RunReport.java
index b7ea62045c5a..de48f4de19b7 100644
--- a/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/RunReport.java
+++ b/tests/performance/report-tool/src/main/java/org/apache/pulsar/tests/performance/report/RunReport.java
@@ -60,13 +60,14 @@ public final class RunReport {
static final String LATENCY_CHART = "latency";
static final String THROUGHPUT_CHART = "throughput";
static final String BACKLOG_CHART = "backlog";
+ static final String REPLAY_CHART = "replay";
static final String TEMPERATURE_CHART = "host-temperature";
static final String FREQUENCY_CHART = "host-frequency";
public static final String RESOLVED_CONFIG = "resolved-config.yaml";
/** The broker's topic stats sampled during a run, one row per topic, subscription and sample. */
public static final String TOPIC_STATS_FILE = "topic-stats.csv";
public static final String TOPIC_STATS_HEADER =
- "epochMillis,topic,subscription,msgBacklog,msgInCounter,msgOutCounter";
+ "epochMillis,topic,subscription,msgBacklog,msgInCounter,msgOutCounter,msgInReplay";
/**
* With offloading, each topic's ledgers and how many of them the brokers had offloaded, when the late applications
* joined (phase {@code join}) and at the run's end (phase {@code end}).
@@ -339,9 +340,18 @@ public final class RunReport {
seconds.stream().mapToDouble(Double::doubleValue).toArray(), cut);
}
- /** The sampled topic stats of a run: one row per sample round. */
+ /**
+ * The sampled topic stats of a run: one row per sample round, by subscription summed over the topics.
+ *
+ * @param replay the entries in each subscription's replay queue, a batch counting once, empty for files sampled
+ * before they were
+ */
record Samples(long[] epochMillis, double[] published, Map<String, double[]> dispatched,
- Map<String, double[]> backlog) {
+ Map<String, double[]> backlog, Map<String, double[]> replay) {
+ Samples(long[] epochMillis, double[] published, Map<String, double[]> dispatched,
+ Map<String, double[]> backlog) {
+ this(epochMillis, published, dispatched, backlog, Map.of());
+ }
}
private RunReport() {
@@ -786,7 +796,8 @@ public final class RunReport {
.append(" ms of its publishing; the time includes opening its pods. Its catch-up rate is the messages"
+ " that it received until then, warmup included, per second since it joined; for an"
+ " application that didn't catch up, it is the rate until its last message. The backlog is"
- + " the sampled topic stats' last sample before the application joined.");
+ + " the sampled topic stats' last sample before the application joined, in entries, so that a"
+ + " batch counts once.");
boolean skipped = late.stream().anyMatch(consumer -> consumer.path("skippedMessages").asLong() > 0);
if (skipped) {
report.append(" Skipped are the messages that the application didn't receive: its subscription was moved"
@@ -1153,7 +1164,8 @@ public final class RunReport {
report.append("\n## Backlog and rates\n\nSampled from the broker's topic stats once per second, or every"
+ " topics / 100 seconds with more than 100 topics; the rates are the deltas of its message counters"
+ " per second between samples, and a sampled maximum is not the exact peak between samples."
- + " The rates are those of the whole seconds within the measurement (0 to ")
+ + " The backlogs are in entries, as the broker counts them: a batch counts once, however many"
+ + " messages it has. The rates are those of the whole seconds within the measurement (0 to ")
.append(String.format(Locale.ROOT, "%.1f", finished))
.append(" s), leaving out its first and last second, where the gateways start and finish. The")
.append(" [sampled topic stats](").append(TOPIC_STATS_FILE).append(") are a CSV file.\n\n")
@@ -1201,10 +1213,51 @@ public final class RunReport {
List<TimeSeriesRenderer.Series> backlog = samples.backlog().entrySet().stream()
.map(entry -> new TimeSeriesRenderer.Series(entry.getKey(), timeline.select(entry.getValue())))
.toList();
- TimeSeriesRenderer.render(runDirectory.resolve(BACKLOG_CHART), "Backlog", "Messages in the backlog",
+ TimeSeriesRenderer.render(runDirectory.resolve(BACKLOG_CHART), "Backlog", "Entries in the backlog",
timeline.seconds(), backlog, finished, timeline.cut(), footer);
report.append("\n![Throughput over time](").append(THROUGHPUT_CHART).append(".svg)\n\n![Backlog over time](")
.append(BACKLOG_CHART).append(".svg)\n");
+ appendReplay(report, runDirectory, samples, seconds, finished, timeline, footer);
+ }
+
+ // The replay queues, of the subscriptions that had entries in replay during the measurement, such as the
+ // Key_Shared entries held back for a consumer without permits and the entries of a closed consumer. The broker
+ // counts the replay queue's positions, entries, so that a batch counts once
+ private static void appendReplay(StringBuilder report, Path runDirectory, Samples samples, double[] seconds,
+ double finished, ChartTimeline timeline, String footer) throws IOException {
+ StringBuilder rows = new StringBuilder();
+ List<TimeSeriesRenderer.Series> series = new ArrayList<>();
+ for (Map.Entry<String, double[]> entry : samples.replay().entrySet()) {
+ double[] window = new double[seconds.length];
+ int count = 0;
+ for (int round = 0; round < seconds.length; round++) {
+ if (seconds[round] >= 0 && seconds[round] <= finished && !Double.isNaN(entry.getValue()[round])) {
+ window[count++] = entry.getValue()[round];
+ }
+ }
+ double[] sorted = Arrays.copyOf(window, count);
+ Arrays.sort(sorted);
+ if (count == 0 || sorted[count - 1] == 0) {
+ continue;
+ }
+ double mean = Arrays.stream(sorted).average().orElse(0);
+ rows.append(String.format(Locale.ROOT, "| `%s` | %,.0f | %,.0f | %,.0f |%n", entry.getKey(), mean,
+ sorted[count / 2], sorted[count - 1]));
+ series.add(new TimeSeriesRenderer.Series(entry.getKey(), timeline.select(entry.getValue())));
+ }
+ if (series.isEmpty()) {
+ return;
+ }
+ TimeSeriesRenderer.render(runDirectory.resolve(REPLAY_CHART), "Replay queue", "Entries in replay",
+ timeline.seconds(), series, finished, timeline.cut(), footer);
+ report.append("\nThe entries in each subscription's replay queue, summed over its topics, sampled during the")
+ .append(" measurement: the entries that the broker holds back to dispatch again, such as a closed")
+ .append(" consumer's unacknowledged entries and, with Key_Shared, the entries of the keys whose")
+ .append(" consumer has no permits. Like the backlog, it is in entries, as is the look-ahead limit")
+ .append(" (`keySharedLookAheadMsgInReplayThresholdPerConsumer`).\n\n")
+ .append("| Subscription | Mean | Median | Sampled maximum |\n")
+ .append("|---|---:|---:|---:|\n").append(rows)
+ .append("\n![Replay queue over time](").append(REPLAY_CHART).append(".svg)\n");
}
private static void appendHost(StringBuilder report, Path runDirectory, HostSamples samples, HostSummary host,
@@ -1520,13 +1573,15 @@ public final class RunReport {
TreeMap<Long, Map<String, Long>> inCounters = new TreeMap<>();
TreeMap<Long, Map<List<String>, long[]>> subscriptions = new TreeMap<>();
List<String> lines = Files.readAllLines(csv);
+ // a file sampled before the replay column existed has none
+ boolean hasReplay = !lines.isEmpty() && lines.get(0).split(",").length > 6;
for (String line : lines.subList(1, lines.size())) {
String[] fields = line.split(",");
long epoch = Long.parseLong(fields[0]);
inCounters.computeIfAbsent(epoch, e -> new LinkedHashMap<>()).put(fields[1], Long.parseLong(fields[4]));
subscriptions.computeIfAbsent(epoch, e -> new HashMap<>())
.put(List.of(fields[1], fields[2]), new long[] {Long.parseLong(fields[3]),
- Long.parseLong(fields[5])});
+ Long.parseLong(fields[5]), fields.length > 6 ? Long.parseLong(fields[6]) : 0});
}
long[] epochs = inCounters.keySet().stream().mapToLong(Long::longValue).toArray();
int rounds = epochs.length;
@@ -1534,11 +1589,16 @@ public final class RunReport {
Arrays.fill(published, Double.NaN);
Map<String, double[]> dispatched = new TreeMap<>();
Map<String, double[]> backlog = new TreeMap<>();
+ Map<String, double[]> replay = new TreeMap<>();
for (int round = 0; round < rounds; round++) {
Map<List<String>, long[]> current = subscriptions.get(epochs[round]);
for (Map.Entry<List<String>, long[]> entry : current.entrySet()) {
double[] values = backlog.computeIfAbsent(entry.getKey().get(1), k -> nanArray(rounds));
values[round] = (Double.isNaN(values[round]) ? 0 : values[round]) + entry.getValue()[0];
+ if (hasReplay) {
+ double[] inReplay = replay.computeIfAbsent(entry.getKey().get(1), k -> nanArray(rounds));
+ inReplay[round] = (Double.isNaN(inReplay[round]) ? 0 : inReplay[round]) + entry.getValue()[2];
+ }
dispatched.computeIfAbsent(entry.getKey().get(1), k -> nanArray(rounds));
}
if (round == 0) {
@@ -1575,7 +1635,7 @@ public final class RunReport {
}
}
}
- return new Samples(epochs, published, dispatched, backlog);
+ return new Samples(epochs, published, dispatched, backlog, replay);
}
private static double[] nanArray(int length) {
diff --git a/tests/performance/report-tool/src/test/java/org/apache/pulsar/tests/performance/report/RunReportTest.java b/tests/performance/report-tool/src/test/java/org/apache/pulsar/tests/performance/report/RunReportTest.java
index 9442fb03127a..dae7988b2074 100644
--- a/tests/performance/report-tool/src/test/java/org/apache/pulsar/tests/performance/report/RunReportTest.java
+++ b/tests/performance/report-tool/src/test/java/org/apache/pulsar/tests/performance/report/RunReportTest.java
@@ -309,9 +309,11 @@ public class RunReportTest {
// sub-0 carries a warmup backlog before the measurement, which its maximum leaves out
csv.append(epoch).append(",persistent://public/default/t-0,sub-0,")
.append(round == 0 ? 900_000 : in[round] - out0[round])
- .append(',').append(in[round]).append(',').append(out0[round]).append('\n');
+ .append(',').append(in[round]).append(',').append(out0[round]).append(",0\n");
+ // sub-1 holds entries in replay while it stalls
csv.append(epoch).append(",persistent://public/default/t-0,sub-1,").append(in[round] - out1[round])
- .append(',').append(in[round]).append(',').append(out1[round]).append('\n');
+ .append(',').append(in[round]).append(',').append(out1[round]).append(',')
+ .append(round == 3 ? 4000 : 0).append('\n');
}
Files.writeString(run.resolve(RunReport.TOPIC_STATS_FILE), csv);
Files.writeString(run.resolve("scenario.yaml"), "extends: base.yaml\n");
@@ -427,6 +429,10 @@ public class RunReportTest {
assertThat(report).contains("| Dispatched msg/s, all subscriptions | 250,000 | 100,000 |");
assertThat(report).contains("| `sub-1` | 100,000 | 2 s | 50,000 |");
assertThat(report).contains("| `sub-0` | 0 | 0 s | 0 |");
+ // Only sub-1 had entries in replay, sampled at most 4,000
+ assertThat(report).containsPattern("\\| `sub-1` \\| [0-9,]+ \\| 0 \\| 4,000 \\|\n");
+ assertThat(report).contains("](replay.svg)");
+ assertThat(run.resolve("replay.svg")).isRegularFile();
// The report shows the SVG charts; each also has a PNG beside it
for (String chart : new String[] {"latency-percentiles.svg", "latency-timeline.svg", "throughput.svg",
"backlog.svg"}) {
@@ -762,6 +768,20 @@ public class RunReportTest {
assertThat(samples.backlog().get("s")[1]).isEqualTo(50.0);
}
+ @Test
+ public void readsTheReplayQueuesWhenSampled() throws IOException {
+ Files.writeString(run.resolve(RunReport.TOPIC_STATS_FILE), RunReport.TOPIC_STATS_HEADER + "\n"
+ + "1000,t-0,a,0,0,0,5\n1000,t-1,a,0,0,0,7\n2000,t-0,a,0,10,10,0\n2000,t-1,a,0,10,10,1\n");
+
+ assertThat(RunReport.readSamples(run.resolve(RunReport.TOPIC_STATS_FILE)).replay().get("a"))
+ .containsExactly(12.0, 1.0);
+
+ // a file sampled before the replay column existed
+ Files.writeString(run.resolve(RunReport.TOPIC_STATS_FILE),
+ "epochMillis,topic,subscription,msgBacklog,msgInCounter,msgOutCounter\n1000,t,s,0,0,0\n");
+ assertThat(RunReport.readSamples(run.resolve(RunReport.TOPIC_STATS_FILE)).replay()).isEmpty();
+ }
+
@Test
public void sumsEachSubscriptionOverItsTopics() throws IOException {
// Two topics consumed by the same two subscriptions; t-1 has no sample in the second round
diff --git a/tests/performance/scenarios/docs/iot-telemetry.md b/tests/performance/scenarios/docs/iot-telemetry.md
index 7e22e88cabcf..97eba4751a26 100644
--- a/tests/performance/scenarios/docs/iot-telemetry.md
+++ b/tests/performance/scenarios/docs/iot-telemetry.md
@@ -32,6 +32,13 @@ application-visible order across Key_Shared hash-range reassignment.
- [`iot-telemetry.yaml`](../iot-telemetry.yaml) is the full topology without churn.
- [`iot-telemetry-restarts.yaml`](../iot-telemetry-restarts.yaml) restarts 10% of each application's
pods every 30 seconds.
+- [`iot-telemetry-key-shared-replay.yaml`](../iot-telemetry-key-shared-replay.yaml) stalls and restarts Key_Shared
+ consumers so that the subscriptions' replay queues hold thousands of entries: 30,000 msg/s with string keys to five
+ applications of 20 pods on four topics, each application pausing three pods for five seconds every 15 seconds and
+ restarting one every 20 seconds. The report's Backlog and rates section shows the replay queues over time.
+- [`iot-telemetry-key-shared-stuck-consumer.yaml`](../iot-telemetry-key-shared-stuck-consumer.yaml) pauses one of
+ each application's five pods for 45 seconds every minute, so that the replay queues reach their look-ahead limit
+ and dispatch stops until the pod resumes: what the broker spends while dispatch is stuck.
- [`iot-telemetry-small.yaml`](../iot-telemetry-small.yaml) keeps the 20-way fanout,
30 topics and 1,000 msg/s rate, but uses 10 gateways and 10 pods per application, with the low-memory
configuration.
@@ -187,6 +194,7 @@ workloads:
size: 64 # bytes, the device's ID, sequence and send time included
devices:
count: 300000
+ keyType: bytes # not set in the base file: bytes, the device's ID as 8 bytes, or string, device-<ID>
gateways:
count: 100
producer: # the gateways' Pulsar clients and producers
@@ -221,8 +229,18 @@ workloads:
podRestarts: # each application restarts this fraction of its pods every intervalSeconds
intervalSeconds: 0
fraction: 0.0
+ podPauses: # not set in the base file: each application pauses this fraction of its pods
+ intervalSeconds: 0 # every intervalSeconds, for pauseMillis
+ fraction: 0.0
+ pauseMillis: 0
```
+`behaviors.podPauses` stalls pods as a garbage collection, a slow downstream or a debugger would: every
+`intervalSeconds`, each application pauses a random `fraction` of its pods' consumers for `pauseMillis`, shorter than
+the interval, and resumes them. A paused consumer stops asking for messages once it has received the ones it already
+asked for, so with Key_Shared the broker holds back its keys' messages in the subscription's replay queue, and
+dispatches the other keys' messages around them, until the replay queue reaches its look-ahead limit.
+
`applications.joinSeconds` has each application join after the measurement starts: a list with a value for each
application, in seconds, where 0 joins at the start, as every application does without the setting. An application
that joins later creates its subscription at the earliest position when the applications start, before the gateways
@@ -251,6 +269,11 @@ applications check each device's order, and with Key_Shared the gateways keep a
section describes. Skipping messages and resetting subscriptions need Key_Shared or Failover, since an application
then expects each device's messages from the first one that it receives.
+`devices.keyType` sets the messages' key. With `bytes`, the default, it's the device's ID as 8 big-endian bytes, which
+the broker receives base64-encoded; with `string`, it's `device-<ID>` as a plain string key, the key type of most
+applications, which Key_Shared hashes as a string. The applications check that each message has its device's key
+either way.
+
`timeoutSeconds` bounds the whole workload: the applications stop waiting for messages after it and the run fails,
the gateways give up waiting for a warmup round or the measurement's start, and the launcher waits for the containers
a minute longer. It has to be at least the time that the warmup rounds, their delays and the measurement need at the
diff --git a/tests/performance/scenarios/iot-telemetry-key-shared-replay.yaml b/tests/performance/scenarios/iot-telemetry-key-shared-replay.yaml
new file mode 100644
index 000000000000..550a451ebe71
--- /dev/null
+++ b/tests/performance/scenarios/iot-telemetry-key-shared-replay.yaml
@@ -0,0 +1,56 @@
+# 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.
+# Key_Shared under stalls and rolling restarts: a large replay queue. Five applications consume the telemetry of
+# 100,000 devices, with string keys, from four topics through 20 pods each. Every 15 seconds each application pauses
+# three of its pods for five seconds, as a garbage collection or a slow downstream would stall them, and every 20
+# seconds it restarts one, as a rolling deployment does. While a pod is paused, the broker holds back its keys'
+# messages in the subscription's replay queue, thousands of entries, and dispatches the other keys' messages around
+# them; a restarted pod's unacknowledged messages go there too. It measures the Key_Shared dispatcher's replay path: the
+# replay queue's bookkeeping, the filtering of its positions on every read, and the replay reads.
+extends: [iot-telemetry-base.yaml, configs/iot-telemetry-high-mem.yaml]
+workloads:
+ iotTelemetry:
+ warmup:
+ seconds: 20
+ measurement:
+ seconds: 180
+ timeoutSeconds: 420
+ rate: 30000
+ payload:
+ size: 128
+ devices:
+ count: 100000
+ keyType: string
+ gateways:
+ count: 20
+ producer:
+ precreate: true
+ topics:
+ count: 4
+ prefix: persistent://public/default/iot-replay-
+ applications:
+ count: 5
+ podsPerApplication: 20
+ subscriptionPrefix: replay-application-
+ behaviors:
+ podPauses:
+ intervalSeconds: 15
+ fraction: 0.15
+ pauseMillis: 5000
+ podRestarts:
+ intervalSeconds: 20
+ fraction: 0.05
diff --git a/tests/performance/scenarios/iot-telemetry-key-shared-stuck-consumer.yaml b/tests/performance/scenarios/iot-telemetry-key-shared-stuck-consumer.yaml
new file mode 100644
index 000000000000..41da8ed9aeb9
--- /dev/null
+++ b/tests/performance/scenarios/iot-telemetry-key-shared-stuck-consumer.yaml
@@ -0,0 +1,37 @@
+# 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.
+# Key_Shared with a stuck consumer: like iot-telemetry-key-shared-replay.yaml, but each application has five pods, and
+# every 60 seconds it pauses one of them for 45 seconds. The paused pod's keys, a fifth of the messages, fill the
+# subscription's replay queue up to its look-ahead limit within seconds; from then on the dispatcher stops reading
+# new messages, so the other pods wait too, until the pod resumes and the replay queue drains. It measures what the
+# broker spends while dispatch is stuck at the limit, and the drain after the pause. The pauses go on after the
+# gateways finish, so the applications drain the backlog in the 15 seconds between them, which takes several minutes.
+extends: iot-telemetry-key-shared-replay.yaml
+workloads:
+ iotTelemetry:
+ rate: 20000
+ timeoutSeconds: 1200
+ applications:
+ podsPerApplication: 5
+ behaviors:
+ podPauses:
+ intervalSeconds: 60
+ fraction: 0.2
+ pauseMillis: 45000
+ podRestarts:
+ intervalSeconds: 0
+ fraction: 0
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 12cdf5f4f9e1..d8cb50d629b5 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
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.SubscriptionType;
/**
@@ -60,8 +61,29 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
public record Payload(int size) {
}
- /** The devices whose telemetry the gateways publish, each with its own key and sequence. */
- public record Devices(int count) {
+ /**
+ * The devices whose telemetry the gateways publish, each with its own key and sequence.
+ *
+ * @param keyType the messages' key: {@code bytes}, the default, the device's ID as 8 big-endian bytes, which the
+ * broker receives base64-encoded, or {@code string}, {@code device-<ID>} as a plain string key,
+ * which Key_Shared hashes as a string
+ */
+ public record Devices(int count, String keyType) {
+ /** The {@code keyType} of binary keys, the default. */
+ public static final String BYTES_KEYS = "bytes";
+ /** The {@code keyType} of string keys. */
+ public static final String STRING_KEYS = "string";
+ static final List<String> KEY_TYPES = List.of(BYTES_KEYS, STRING_KEYS);
+
+ public Devices {
+ if (keyType == null) {
+ keyType = BYTES_KEYS;
+ }
+ }
+
+ public Devices(int count) {
+ this(count, null);
+ }
}
/**
@@ -195,11 +217,16 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
}
/** What happens to the simulated system during the run, beyond the traffic. */
- public record Behaviors(PodRestarts podRestarts) {
- static final Behaviors NONE = new Behaviors(PodRestarts.NONE);
+ public record Behaviors(PodRestarts podRestarts, PodPauses podPauses) {
+ static final Behaviors NONE = new Behaviors(PodRestarts.NONE, PodPauses.NONE);
public Behaviors {
podRestarts = podRestarts != null ? podRestarts : PodRestarts.NONE;
+ podPauses = podPauses != null ? podPauses : PodPauses.NONE;
+ }
+
+ public Behaviors(PodRestarts podRestarts) {
+ this(podRestarts, null);
}
}
@@ -217,6 +244,23 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
}
}
+ /**
+ * Each application pauses some of its pods periodically, as a pod stalled by a garbage collection, a slow
+ * downstream or a debugger would: a paused pod's consumer stops asking for messages, so the broker holds back its
+ * keys' messages, which a Key_Shared subscription keeps in its replay queue, until it resumes.
+ *
+ * @param intervalSeconds how often each application pauses pods; 0 never
+ * @param fraction the fraction of an application's pods that each pause stops
+ * @param pauseMillis how long the pods stay paused, shorter than the interval
+ */
+ public record PodPauses(int intervalSeconds, double fraction, int pauseMillis) {
+ static final PodPauses NONE = new PodPauses(0, 0, 0);
+
+ boolean enabled() {
+ return intervalSeconds > 0 && fraction > 0 && pauseMillis > 0;
+ }
+ }
+
public IotScenario {
warmup = warmup != null ? warmup : Warmup.NONE;
behaviors = behaviors != null ? behaviors : Behaviors.NONE;
@@ -249,6 +293,7 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
Producer producer = gateways.producer();
Client client = applications.client();
PodRestarts podRestarts = behaviors.podRestarts();
+ PodPauses podPauses = behaviors.podPauses();
require(durationSeconds >= 1, "measurement.seconds must be at least 1");
require(warmupSeconds >= 0 && warmupMessages >= 0 && warmupRoundDelaySeconds >= 0,
"warmup.seconds, warmup.messages and warmup.roundDelaySeconds must not be negative");
@@ -263,6 +308,8 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
&& applications.podsPerApplication() >= 1,
"devices.count, gateways.count, topics.count, applications.count and "
+ "applications.podsPerApplication must be at least 1");
+ require(Devices.KEY_TYPES.contains(devices.keyType()),
+ "devices.keyType must be one of " + Devices.KEY_TYPES + ", not " + devices.keyType());
require(producer.ioThreads() >= 1 && producer.listenerThreads() >= 1 && client.ioThreads() >= 1
&& client.listenerThreads() >= 1,
"The gateways' and the applications' ioThreads and listenerThreads must be at least 1");
@@ -351,6 +398,13 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
}
require(podRestarts.intervalSeconds() >= 0 && podRestarts.fraction() >= 0 && podRestarts.fraction() <= 1,
"behaviors.podRestarts.intervalSeconds must not be negative, and fraction must be from 0 to 1");
+ require(podPauses.intervalSeconds() >= 0 && podPauses.fraction() >= 0 && podPauses.fraction() <= 1
+ && podPauses.pauseMillis() >= 0,
+ "behaviors.podPauses.intervalSeconds and pauseMillis must not be negative, and fraction must be from 0"
+ + " to 1");
+ require(!podPauses.enabled()
+ || podPauses.pauseMillis() < TimeUnit.SECONDS.toMillis(podPauses.intervalSeconds()),
+ "behaviors.podPauses.pauseMillis must be shorter than intervalSeconds");
}
private static void require(boolean condition, String message) {
@@ -383,6 +437,11 @@ public record IotScenario(String serviceUrl, Warmup warmup, Measurement measurem
return devices.count();
}
+ /** Whether the gateways key the messages with strings rather than with bytes. */
+ public boolean stringKeys() {
+ return Devices.STRING_KEYS.equals(devices.keyType());
+ }
+
public int gatewayCount() {
return gateways.count();
}
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 113f1469a9f7..1ede428e6b13 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
@@ -20,11 +20,11 @@ package org.apache.pulsar.tests.performance.tools;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
@@ -45,6 +45,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientSharedResources;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
@@ -279,6 +280,7 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
private final AtomicBoolean stopping = new AtomicBoolean();
private final AtomicReference<Throwable> restartFailure = new AtomicReference<>();
private volatile Thread restarter;
+ private volatile Thread pauser;
private int nextWarmupRound = 1;
private volatile boolean finished;
@@ -408,6 +410,11 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
restarter = thread;
thread.start();
}
+ if (scenario.behaviors().podPauses().enabled() && !stopping.get()) {
+ Thread thread = new Thread(this::pausePods, "iot-client-pauser-" + index);
+ pauser = thread;
+ thread.start();
+ }
}
void checkRestarts() {
@@ -472,6 +479,7 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
// so that they don't open pods after they're closed, or use the shared resources after they are
interruptAndAwait(joiner);
interruptAndAwait(restarter);
+ interruptAndAwait(pauser);
closePods();
}
@@ -479,6 +487,7 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
stopping.set(true);
interruptAndAwait(joiner);
interruptAndAwait(restarter);
+ interruptAndAwait(pauser);
}
/**
@@ -536,9 +545,7 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
listenersInFlight.incrementAndGet();
try {
TelemetryMessage.Decoded decoded = TelemetryMessage.decode(message.getData());
- byte[] key = message.getKeyBytes();
- if (key == null || key.length != Long.BYTES
- || ByteBuffer.wrap(key).getLong() != decoded.deviceId()) {
+ if (!TelemetryMessage.hasKey(message, decoded.deviceId(), scenario.stringKeys())) {
throw new IllegalArgumentException(
"Telemetry key does not match payload device ID");
}
@@ -598,6 +605,36 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
}
}
}
+
+ // Pauses random pods' consumers for the pause's time every interval, and resumes them then, also when the
+ // pauses stop. A pod that a restart replaces meanwhile is resumed, closed, harmlessly.
+ private void pausePods() {
+ IotScenario.PodPauses podPauses = scenario.behaviors().podPauses();
+ int pauseCount = Math.max(1, (int) Math.ceil(scenario.podsPerApplication() * podPauses.fraction()));
+ List<Consumer<byte[]>> paused = new ArrayList<>(pauseCount);
+ try {
+ while (!stopping.get()) {
+ Thread.sleep(TimeUnit.SECONDS.toMillis(podPauses.intervalSeconds()) - podPauses.pauseMillis());
+ synchronized (pods) {
+ List<ClientAndConsumer> candidates = new ArrayList<>(pods);
+ Collections.shuffle(candidates);
+ for (ClientAndConsumer pod : candidates.subList(0, Math.min(pauseCount, candidates.size()))) {
+ pod.consumer().pause();
+ paused.add(pod.consumer());
+ }
+ }
+ Thread.sleep(podPauses.pauseMillis());
+ paused.forEach(Consumer::resume);
+ paused.clear();
+ }
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ } catch (RuntimeException error) {
+ restartFailure.compareAndSet(null, error);
+ } finally {
+ paused.forEach(Consumer::resume);
+ }
+ }
}
private record ClientAndConsumer(PulsarClient client, Consumer<byte[]> consumer) implements AutoCloseable {
@@ -610,14 +647,27 @@ final class TelemetryConsumer extends PerformanceTool.ScenarioCommand {
}
}
- /** Closes the consumer, then the client, also when closing the consumer failed. */
+ /**
+ * Closes the consumer, then the client, also when closing the consumer failed. A pod that a restart closed and
+ * couldn't replace, as when the application stops during the restart, is already closed, which isn't a
+ * failure.
+ */
CompletableFuture<Void> closeAsync() {
- return consumer.closeAsync().handle((ignored, consumerFailure) -> consumerFailure)
- .thenCompose(consumerFailure -> client.closeAsync().thenRun(() -> {
- if (consumerFailure != null) {
- throw new CompletionException(consumerFailure);
- }
- }));
+ return consumer.closeAsync().handle((ignored, consumerFailure) -> unlessAlreadyClosed(consumerFailure))
+ .thenCompose(consumerFailure -> client.closeAsync()
+ .exceptionallyCompose(failure -> unlessAlreadyClosed(failure) == null
+ ? CompletableFuture.completedFuture(null) : CompletableFuture.failedFuture(failure))
+ .thenRun(() -> {
+ if (consumerFailure != null) {
+ throw new CompletionException(consumerFailure);
+ }
+ }));
+ }
+
+ private static Throwable unlessAlreadyClosed(Throwable failure) {
+ Throwable cause = failure instanceof CompletionException && failure.getCause() != null
+ ? failure.getCause() : failure;
+ return cause instanceof PulsarClientException.AlreadyClosedException ? null : failure;
}
}
}
diff --git a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryMessage.java b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryMessage.java
index f4708d72e9db..f7ea99c5af71 100644
--- a/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryMessage.java
+++ b/tests/performance/tools/src/main/java/org/apache/pulsar/tests/performance/tools/TelemetryMessage.java
@@ -19,6 +19,8 @@
package org.apache.pulsar.tests.performance.tools;
import java.nio.ByteBuffer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
final class TelemetryMessage {
static final int HEADER_BYTES = Long.BYTES * 4;
@@ -43,6 +45,26 @@ final class TelemetryMessage {
(buffer.getLong() & MEASUREMENT_FLAG) != 0);
}
+ /** Keys the device's message: its ID as 8 big-endian bytes, or {@code device-<ID>} with string keys. */
+ static TypedMessageBuilder<byte[]> key(TypedMessageBuilder<byte[]> message, long deviceId, boolean stringKeys) {
+ return stringKeys ? message.key(stringKey(deviceId))
+ : message.keyBytes(ByteBuffer.allocate(Long.BYTES).putLong(deviceId).array());
+ }
+
+ /** Whether the message has the key that {@link #key} gives the device's messages. */
+ static boolean hasKey(Message<?> message, long deviceId, boolean stringKeys) {
+ if (stringKeys) {
+ // a string key reaches the consumer as it was sent, not base64-encoded
+ return !message.hasBase64EncodedKey() && stringKey(deviceId).equals(message.getKey());
+ }
+ byte[] key = message.getKeyBytes();
+ return key != null && key.length == Long.BYTES && ByteBuffer.wrap(key).getLong() == deviceId;
+ }
+
+ private static String stringKey(long deviceId) {
+ return "device-" + deviceId;
+ }
+
private TelemetryMessage() {
}
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 dfdf5270d63f..c7f559114571 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
@@ -20,7 +20,6 @@ package org.apache.pulsar.tests.performance.tools;
import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
-import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
@@ -68,6 +67,7 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
@Override
public Integer call() throws Exception {
IotScenario scenario = scenario();
+ boolean stringKeys = scenario.stringKeys();
Files.createDirectories(output);
long[] deviceSequences = new long[scenario.deviceCount()];
long[] producerSequences = new long[Math.multiplyExact(scenario.gatewayCount(), scenario.topicCount())];
@@ -178,13 +178,11 @@ final class TelemetryProducer extends PerformanceTool.ScenarioCommand {
}
long deviceSequence = deviceSequences[device]++;
long producerSequence = producerSequences[producerIndex]++;
- byte[] key = ByteBuffer.allocate(Long.BYTES).putLong(device).array();
byte[] payload = TelemetryMessage.encode(device, deviceSequence, measurementMessage,
scenario.payloadBytes());
int completedDevice = device;
long sendStartedNanos = System.nanoTime();
- producer.newMessage()
- .keyBytes(key)
+ TelemetryMessage.key(producer.newMessage(), device, stringKeys)
.sequenceId(producerSequence)
.value(payload)
.sendAsync()
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 044f76ced5ef..e5b72fbf3dda 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
@@ -233,6 +233,53 @@ public class IotScenarioTest {
assertThat(scenario(fromYaml).subscriptionType()).isEqualTo(SubscriptionType.Key_Shared);
}
+ @Test
+ public void validatesThePodPauses() {
+ assertThat(withBehaviors(null).behaviors().podPauses().enabled()).isFalse();
+ assertThat(withBehaviors(new IotScenario.Behaviors(null, new IotScenario.PodPauses(15, 0.1, 5000)))
+ .behaviors().podPauses().enabled()).isTrue();
+ assertThatThrownBy(() -> withBehaviors(new IotScenario.Behaviors(null,
+ new IotScenario.PodPauses(5, 0.1, 5000))))
+ .hasMessageContaining("behaviors.podPauses.pauseMillis must be shorter than intervalSeconds");
+ assertThatThrownBy(() -> withBehaviors(new IotScenario.Behaviors(null,
+ new IotScenario.PodPauses(15, 1.5, 5000))))
+ .hasMessageContaining("fraction must be from 0 to 1");
+ // a scenario file with restarts only has no pauses
+ IotScenario.Behaviors fromYaml = new YamlScenarioLoader().mapper().convertValue(
+ Map.of("podRestarts", Map.of("intervalSeconds", 30, "fraction", 0.1)), IotScenario.Behaviors.class);
+ assertThat(withBehaviors(fromYaml).behaviors().podPauses().enabled()).isFalse();
+ }
+
+ private static IotScenario withBehaviors(IotScenario.Behaviors behaviors) {
+ return new IotScenario("pulsar://localhost:6650", new IotScenario.Warmup(0, 0, 1, 0),
+ new IotScenario.Measurement(120, 1_000), 0, new IotScenario.Payload(64),
+ new IotScenario.Devices(1_000),
+ new IotScenario.Gateways(10, new IotScenario.Producer(2, 2, 100, true, true, null), null),
+ new IotScenario.Topics(2, "persistent://public/default/iot-"),
+ new IotScenario.Applications(1, 2, "app-", new IotScenario.Client(2, 2), null), behaviors, 300);
+ }
+
+ @Test
+ public void choosesTheKeyType() {
+ assertThat(withDevices(new IotScenario.Devices(1_000)).stringKeys()).isFalse();
+ assertThat(withDevices(new IotScenario.Devices(1_000, "bytes")).stringKeys()).isFalse();
+ assertThat(withDevices(new IotScenario.Devices(1_000, "string")).stringKeys()).isTrue();
+ assertThatThrownBy(() -> withDevices(new IotScenario.Devices(1_000, "long")))
+ .hasMessageContaining("devices.keyType must be one of [bytes, string], not long");
+ // a scenario file without the setting keeps binary keys
+ IotScenario.Devices fromYaml = new YamlScenarioLoader().mapper().convertValue(Map.of("count", 1_000),
+ IotScenario.Devices.class);
+ assertThat(withDevices(fromYaml).stringKeys()).isFalse();
+ }
+
+ private static IotScenario withDevices(IotScenario.Devices devices) {
+ return new IotScenario("pulsar://localhost:6650", new IotScenario.Warmup(0, 0, 1, 0),
+ new IotScenario.Measurement(120, 1_000), 0, new IotScenario.Payload(64), devices,
+ new IotScenario.Gateways(10, new IotScenario.Producer(2, 2, 100, true, true, null), null),
+ new IotScenario.Topics(2, "persistent://public/default/iot-"),
+ new IotScenario.Applications(1, 2, "app-", new IotScenario.Client(2, 2), null), null, 300);
+ }
+
@Test
public void rejectsInvalidJoinSettings() {
assertThatThrownBy(() -> scenario(new IotScenario.Applications(3, 2, "app-", new IotScenario.Client(2, 2),
diff --git a/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/TelemetryMessageTest.java b/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/TelemetryMessageTest.java
index 0c9370de2bea..c1960af66fb1 100644
--- a/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/TelemetryMessageTest.java
+++ b/tests/performance/tools/src/test/java/org/apache/pulsar/tests/performance/tools/TelemetryMessageTest.java
@@ -19,6 +19,13 @@
package org.apache.pulsar.tests.performance.tools;
import static org.assertj.core.api.Assertions.assertThat;
+import java.lang.reflect.Proxy;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.testng.annotations.Test;
public class TelemetryMessageTest {
@@ -36,4 +43,40 @@ public class TelemetryMessageTest {
assertThat(measurement.sequence()).isEqualTo(8);
assertThat(measurement.measurement()).isTrue();
}
+
+ @Test
+ public void matchesTheKeyOfTheDevice() {
+ for (boolean stringKeys : new boolean[] {false, true}) {
+ Map<String, Object> key = new HashMap<>();
+ TelemetryMessage.key(builder(key), 42, stringKeys);
+ assertThat(key).containsOnlyKeys(stringKeys ? "key" : "keyBytes");
+ Message<?> message = received(key);
+ assertThat(TelemetryMessage.hasKey(message, 42, stringKeys)).isTrue();
+ assertThat(TelemetryMessage.hasKey(message, 43, stringKeys)).isFalse();
+ assertThat(TelemetryMessage.hasKey(message, 42, !stringKeys)).isFalse();
+ }
+ }
+
+ // A message builder that records the key that it's given
+ @SuppressWarnings("unchecked")
+ private static TypedMessageBuilder<byte[]> builder(Map<String, Object> key) {
+ return (TypedMessageBuilder<byte[]>) Proxy.newProxyInstance(TypedMessageBuilder.class.getClassLoader(),
+ new Class<?>[] {TypedMessageBuilder.class}, (proxy, method, args) -> {
+ key.put(method.getName(), args[0]);
+ return proxy;
+ });
+ }
+
+ // The message that a consumer receives with the recorded key, where a binary key arrives base64-encoded
+ private static Message<?> received(Map<String, Object> key) {
+ byte[] keyBytes = (byte[]) key.get("keyBytes");
+ String stringKey = keyBytes != null ? Base64.getEncoder().encodeToString(keyBytes) : (String) key.get("key");
+ return (Message<?>) Proxy.newProxyInstance(Message.class.getClassLoader(), new Class<?>[] {Message.class},
+ (proxy, method, args) -> switch (method.getName()) {
+ case "hasBase64EncodedKey" -> keyBytes != null;
+ case "getKey" -> stringKey;
+ case "getKeyBytes" -> keyBytes != null ? keyBytes : stringKey.getBytes(StandardCharsets.UTF_8);
+ default -> throw new UnsupportedOperationException(method.getName());
+ });
+ }
}