You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/05 01:54:15 UTC
(pulsar) branch master updated: [improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new bc1785dcd867 [improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
bc1785dcd867 is described below
commit bc1785dcd867c8e9a03310b84cab5a0a51d14ef5
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Sep 5 04:54:08 2026 +0300
[improve][cli] pulsar-perf: cut latency histogram memory ~100x, fix range units and static stats state (#26466)
---
.../proxy/socket/client/PerformanceClient.java | 23 +++++---
.../socket/client/SimpleTestProducerSocket.java | 13 ++++-
.../pulsar/testclient/ManagedLedgerWriter.java | 25 +++++---
.../apache/pulsar/testclient/PerfClientUtils.java | 17 ++++++
.../pulsar/testclient/PerformanceConsumer.java | 53 +++++++----------
.../pulsar/testclient/PerformanceProducer.java | 54 +++++++----------
.../pulsar/testclient/PerformanceReader.java | 26 +++++----
.../pulsar/testclient/PerformanceTransaction.java | 67 ++++++++++------------
.../pulsar/testclient/PerfClientUtilsTest.java | 22 +++++++
9 files changed, 166 insertions(+), 134 deletions(-)
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
index 5a78a8e69271..ca601e72b124 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.proxy.socket.client;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.google.common.util.concurrent.RateLimiter;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.io.FileInputStream;
@@ -44,6 +45,7 @@ import java.util.concurrent.atomic.LongAdder;
import lombok.CustomLog;
import org.HdrHistogram.Histogram;
import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
@@ -66,10 +68,13 @@ import picocli.CommandLine.Parameters;
@CustomLog
public class PerformanceClient extends CmdBase {
- private static final LongAdder messagesSent = new LongAdder();
- private static final LongAdder bytesSent = new LongAdder();
- private static final LongAdder totalMessagesSent = new LongAdder();
- private static final LongAdder totalBytesSent = new LongAdder();
+ private final LongAdder messagesSent = new LongAdder();
+ private final LongAdder bytesSent = new LongAdder();
+ private final LongAdder totalMessagesSent = new LongAdder();
+ private final LongAdder totalBytesSent = new LongAdder();
+
+ private final Recorder recorder =
+ new Recorder(SimpleTestProducerSocket.MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
private static IMessageFormatter messageFormatter = null;
@Option(names = { "-cf", "--conf-file" }, description = "Configuration file")
@@ -261,7 +266,7 @@ public class PerformanceClient extends CmdBase {
}
}
- SimpleTestProducerSocket produceSocket = new SimpleTestProducerSocket();
+ SimpleTestProducerSocket produceSocket = new SimpleTestProducerSocket(recorder);
try {
produceClient.start();
@@ -373,7 +378,7 @@ public class PerformanceClient extends CmdBase {
double rate = messagesSent.sumThenReset() / elapsed;
double throughput = bytesSent.sumThenReset() / elapsed / 1024 / 1024 * 8;
- reportHistogram = SimpleTestProducerSocket.recorder.getIntervalHistogram(reportHistogram);
+ reportHistogram = recorder.getIntervalHistogram(reportHistogram);
log.infof("Throughput produced: %7d msg --- %8.1f msg/s --- %8.1f Mbit/s"
+ " --- Latency: mean: %7.3f ms - med: %7.3f ms"
@@ -444,7 +449,7 @@ public class PerformanceClient extends CmdBase {
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesSent.sum() / elapsed;
double throughput = totalBytesSent.sum() / elapsed / 1024 / 1024 * 8;
@@ -452,8 +457,8 @@ public class PerformanceClient extends CmdBase {
totalMessagesSent.sum(), rate, throughput);
}
- private static void printAggregatedStats() {
- Histogram reportHistogram = SimpleTestProducerSocket.recorder.getIntervalHistogram();
+ private void printAggregatedStats() {
+ Histogram reportHistogram = recorder.getIntervalHistogram();
log.infof("Aggregated latency stats --- Latency: mean: %7.3f ms"
+ " - med: %d - 95pct: %d - 99pct: %d"
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java
index e5380b2fed1e..0f7613a6bdb1 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/SimpleTestProducerSocket.java
@@ -41,14 +41,21 @@ import org.eclipse.jetty.websocket.api.annotations.WebSocket;
@WebSocket
@CustomLog
public class SimpleTestProducerSocket {
- public static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
+ // Latencies are recorded in microseconds; values above this ceiling are clamped rather than
+ // rejected, since HdrHistogram throws on an out-of-range value. Package-private so that
+ // PerformanceClient, which owns the shared Recorder, can size it consistently.
+ static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
+
+ // Shared by every socket of one PerformanceClient run so that latencies aggregate.
+ private final Recorder recorder;
private final CountDownLatch closeLatch;
private volatile Session session;
private ConcurrentHashMap<String, Long> startTimeMap = new ConcurrentHashMap<>();
private static final String CONTEXT = "context";
- public SimpleTestProducerSocket() {
+ public SimpleTestProducerSocket(Recorder recorder) {
+ this.recorder = recorder;
this.closeLatch = new CountDownLatch(2);
}
@@ -78,7 +85,7 @@ public class SimpleTestProducerSocket {
startTime = startTimeMap.get(json.get(CONTEXT).getAsString());
}
long latencyNs = endTimeNs - startTime;
- recorder.recordValue(NANOSECONDS.toMicros(latencyNs));
+ recorder.recordValue(Math.min(NANOSECONDS.toMicros(latencyNs), MAX_LATENCY_MICROS));
}
public Session getSession() {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
index 12016b608d7c..249551686923 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.util.concurrent.RateLimiter;
@@ -67,13 +68,18 @@ import picocli.CommandLine.Option;
@CustomLog
public class ManagedLedgerWriter extends CmdBase{
- private static final LongAdder messagesSent = new LongAdder();
- private static final LongAdder bytesSent = new LongAdder();
- private static final LongAdder totalMessagesSent = new LongAdder();
- private static final LongAdder totalBytesSent = new LongAdder();
+ private final LongAdder messagesSent = new LongAdder();
+ private final LongAdder bytesSent = new LongAdder();
+ private final LongAdder totalMessagesSent = new LongAdder();
+ private final LongAdder totalBytesSent = new LongAdder();
- private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
- private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
+ // Latencies are recorded in microseconds. A managed-ledger write slower than this means the
+ // benchmark is broken rather than slow, so values are clamped instead of widening the range.
+ private static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
+
+ private final Recorder recorder = new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
@@ -251,7 +257,8 @@ public class ManagedLedgerWriter extends CmdBase{
totalMessagesSent.increment();
totalBytesSent.add(payloadData.length);
- long latencyMicros = NANOSECONDS.toMicros(System.nanoTime() - sendTime);
+ long latencyMicros = Math.min(
+ NANOSECONDS.toMicros(System.nanoTime() - sendTime), MAX_LATENCY_MICROS);
recorder.recordValue(latencyMicros);
cumulativeRecorder.recordValue(latencyMicros);
@@ -399,7 +406,7 @@ public class ManagedLedgerWriter extends CmdBase{
return map;
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesSent.sum() / elapsed;
double throughput = totalBytesSent.sum() / elapsed / 1024 / 1024 * 8;
@@ -410,7 +417,7 @@ public class ManagedLedgerWriter extends CmdBase{
.log("Aggregated throughput stats --- records sent --- msg/s --- Mbit/s");
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.info()
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java
index 81484e2788b6..3276bf95566f 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerfClientUtils.java
@@ -51,6 +51,23 @@ import org.apache.pulsar.tls.TlsPolicy;
@UtilityClass
public class PerfClientUtils {
+ /**
+ * Number of significant decimal digits kept by the perf clients' latency histograms.
+ *
+ * <p>HdrHistogram sizes a fixed-range histogram's counts array proportionally to
+ * {@code 2^ceil(log2(2 * 10^digits))}, so every extra digit multiplies the allocation by
+ * roughly 10. At 5 digits (HdrHistogram's maximum) a single {@code Recorder} over the
+ * ranges used here costs 11-22 MB, and since these recorders are static fields the JVM
+ * pays for every perf subcommand, not just the one being run.
+ *
+ * <p>3 digits bounds the error of a reported percentile at 0.1%. That is far below the
+ * run-to-run variance of a benchmark, and finer than the reports resolve anyway: the
+ * microsecond-based tools print milliseconds with {@code %.3f}, and the millisecond-based
+ * ones print whole milliseconds with {@code %d}. It is also HdrHistogram's own recommended
+ * default.
+ */
+ public static final int LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS = 3;
+
private static volatile Consumer<Integer> exitProcedure = System::exit;
public static void setExitProcedure(Consumer<Integer> exitProcedure) {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
index 448e668ffc27..056bc87e3420 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.util.concurrent.RateLimiter;
@@ -90,26 +91,27 @@ public class PerformanceConsumer extends PerformanceTopicListArguments{
Stream
}
- private static final LongAdder messagesReceived = new LongAdder();
- private static final LongAdder bytesReceived = new LongAdder();
+ private final LongAdder messagesReceived = new LongAdder();
+ private final LongAdder bytesReceived = new LongAdder();
- private static final LongAdder totalMessagesReceived = new LongAdder();
- private static final LongAdder totalBytesReceived = new LongAdder();
+ private final LongAdder totalMessagesReceived = new LongAdder();
+ private final LongAdder totalBytesReceived = new LongAdder();
- private static final LongAdder totalNumTxnOpenFail = new LongAdder();
- private static final LongAdder totalNumTxnOpenSuccess = new LongAdder();
+ private final LongAdder totalNumTxnOpenFail = new LongAdder();
+ private final LongAdder totalNumTxnOpenSuccess = new LongAdder();
- private static final LongAdder totalMessageAck = new LongAdder();
- private static final LongAdder totalMessageAckFailed = new LongAdder();
- private static final LongAdder messageAck = new LongAdder();
+ private final LongAdder totalMessageAck = new LongAdder();
+ private final LongAdder totalMessageAckFailed = new LongAdder();
+ private final LongAdder messageAck = new LongAdder();
- private static final LongAdder totalEndTxnOpFailNum = new LongAdder();
- private static final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
- private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private final LongAdder totalEndTxnOpFailNum = new LongAdder();
+ private final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
+ private final LongAdder numTxnOpSuccess = new LongAdder();
- private static final long MAX_LATENCY = TimeUnit.DAYS.toMillis(10);
- private static final Recorder recorder = new Recorder(MAX_LATENCY, 5);
- private static final Recorder cumulativeRecorder = new Recorder(MAX_LATENCY, 5);
+ private static final long MAX_LATENCY_MILLIS = TimeUnit.DAYS.toMillis(10);
+ private final Recorder recorder = new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = { "-n", "--num-consumers" }, description = "Number of consumers (per subscription), only "
+ "one consumer is allowed when subscriptionType is Exclusive",
@@ -247,21 +249,6 @@ public class PerformanceConsumer extends PerformanceTopicListArguments{
}
@Override
public void run() throws Exception {
- // Reset static counters to avoid stale state from previous runs in the same JVM
- messagesReceived.reset();
- bytesReceived.reset();
- totalMessagesReceived.reset();
- totalBytesReceived.reset();
- totalNumTxnOpenFail.reset();
- totalNumTxnOpenSuccess.reset();
- totalMessageAck.reset();
- totalMessageAckFailed.reset();
- messageAck.reset();
- totalEndTxnOpFailNum.reset();
- totalEndTxnOpSuccessNum.reset();
- numTxnOpSuccess.reset();
- recorder.reset();
- cumulativeRecorder.reset();
// Dump config variables
PerfClientUtils.printJVMInformation(log);
@@ -515,8 +502,8 @@ public class PerformanceConsumer extends PerformanceTopicListArguments{
long latencyMillis = System.currentTimeMillis() - msg.publishTime().toEpochMilli();
if (latencyMillis >= 0) {
- if (latencyMillis >= MAX_LATENCY) {
- latencyMillis = MAX_LATENCY;
+ if (latencyMillis >= MAX_LATENCY_MILLIS) {
+ latencyMillis = MAX_LATENCY_MILLIS;
}
recorder.recordValue(latencyMillis);
cumulativeRecorder.recordValue(latencyMillis);
@@ -749,7 +736,7 @@ public class PerformanceConsumer extends PerformanceTopicListArguments{
totalMessagesReceived.sum(), rate, throughput, rateAck, totalnumMessageAckFailed);
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.infof("Aggregated latency stats --- Latency: mean: %.3f ms"
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
index 05b82a7e156d..47ae0d45229e 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -24,6 +24,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_BATCHING_MAX_MESSAGES;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.collect.Range;
@@ -89,22 +90,28 @@ import picocli.CommandLine.TypeConversionException;
@Command(name = "produce", description = "Test pulsar producer performance.")
@CustomLog
public class PerformanceProducer extends PerformanceTopicListArguments{
- private static final LongAdder messagesSent = new LongAdder();
- private static final LongAdder messagesFailed = new LongAdder();
- private static final LongAdder bytesSent = new LongAdder();
+ private final LongAdder messagesSent = new LongAdder();
+ private final LongAdder messagesFailed = new LongAdder();
+ private final LongAdder bytesSent = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
- private static final LongAdder totalMessagesSent = new LongAdder();
- private static final LongAdder totalBytesSent = new LongAdder();
+ private final LongAdder totalMessagesSent = new LongAdder();
+ private final LongAdder totalBytesSent = new LongAdder();
- private static final Recorder recorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static final Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ // Publish latencies are recorded in microseconds. A send slower than this means the benchmark is
+ // broken rather than slow, so values are clamped to keep HdrHistogram in range.
+ private static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
- private static final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
- private static final LongAdder totalEndTxnOpFailNum = new LongAdder();
- private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private final Recorder recorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+
+ private final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
+ private final LongAdder totalEndTxnOpFailNum = new LongAdder();
+ private final LongAdder numTxnOpSuccess = new LongAdder();
private static IMessageFormatter messageFormatter = null;
@@ -256,19 +263,6 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
@Override
public void run() throws Exception {
- // Reset static counters to avoid stale state from previous runs in the same JVM
- messagesSent.reset();
- messagesFailed.reset();
- bytesSent.reset();
- totalNumTxnOpenTxnFail.reset();
- totalNumTxnOpenTxnSuccess.reset();
- totalMessagesSent.reset();
- totalBytesSent.reset();
- totalEndTxnOpSuccessNum.reset();
- totalEndTxnOpFailNum.reset();
- numTxnOpSuccess.reset();
- recorder.reset();
- cumulativeRecorder.reset();
// Dump config variables
PerfClientUtils.printJVMInformation(log);
@@ -679,17 +673,13 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
long now = System.nanoTime();
if (now > warmupEndTime) {
- long latencyMicros = NANOSECONDS.toMicros(now - sendTime);
+ long latencyMicros =
+ Math.min(NANOSECONDS.toMicros(now - sendTime), MAX_LATENCY_MICROS);
recorder.recordValue(latencyMicros);
cumulativeRecorder.recordValue(latencyMicros);
}
}).exceptionally(ex -> {
- // Ignore the exception of recorder since a very large latencyMicros will lead
- // ArrayIndexOutOfBoundsException in AbstractHistogram
Throwable cause = FutureUtil.unwrapCompletionException(ex);
- if (cause instanceof ArrayIndexOutOfBoundsException) {
- return null;
- }
// Ignore the exception when the producer is closed
if (cause instanceof PulsarClientException.AlreadyClosedException) {
return null;
@@ -819,7 +809,7 @@ public class PerformanceProducer extends PerformanceTopicListArguments{
totalMessagesSent.sum(), rate, throughput);
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.infof("Aggregated latency stats --- Latency: mean: %7.3f ms"
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
index a443359be7ab..e232b83f4d95 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.testclient;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import static org.apache.pulsar.testclient.PerfClientUtils.addShutdownHook;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
@@ -50,14 +51,17 @@ import picocli.CommandLine.Option;
@Command(name = "read", description = "Test pulsar reader performance.")
@CustomLog
public class PerformanceReader extends PerformanceTopicListArguments {
- private static final LongAdder messagesReceived = new LongAdder();
- private static final LongAdder bytesReceived = new LongAdder();
+ private final LongAdder messagesReceived = new LongAdder();
+ private final LongAdder bytesReceived = new LongAdder();
- private static final LongAdder totalMessagesReceived = new LongAdder();
- private static final LongAdder totalBytesReceived = new LongAdder();
+ private final LongAdder totalMessagesReceived = new LongAdder();
+ private final LongAdder totalBytesReceived = new LongAdder();
- private static Recorder recorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
- private static Recorder cumulativeRecorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
+ private static final long MAX_LATENCY_MILLIS = TimeUnit.DAYS.toMillis(10);
+
+ private final Recorder recorder = new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder cumulativeRecorder =
+ new Recorder(MAX_LATENCY_MILLIS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = {"-r", "--rate"}, description = "Simulate a slow message reader (rate in msg/s)")
public double rate = 0;
@@ -256,13 +260,15 @@ public class PerformanceReader extends PerformanceTopicListArguments {
long latencyMillis = System.currentTimeMillis() - msg.publishTime().toEpochMilli();
if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
- cumulativeRecorder.recordValue(latencyMillis);
+ // Reading a backlog older than the histogram range must not blow up the read loop.
+ long clampedLatencyMillis = Math.min(latencyMillis, MAX_LATENCY_MILLIS);
+ recorder.recordValue(clampedLatencyMillis);
+ cumulativeRecorder.recordValue(clampedLatencyMillis);
}
}
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesReceived.sum() / elapsed;
double throughput = totalBytesReceived.sum() / elapsed * 8 / 1024 / 1024;
@@ -270,7 +276,7 @@ public class PerformanceReader extends PerformanceTopicListArguments {
totalMessagesReceived.sum(), rate, throughput);
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportHistogram = cumulativeRecorder.getIntervalHistogram();
log.infof("Aggregated latency stats --- Latency: mean: %.3f ms"
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
index 7a84f7894a3b..86843e901eac 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.testclient;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static org.apache.pulsar.testclient.PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS;
import static org.apache.pulsar.testclient.PerfClientUtils.addShutdownHook;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
@@ -77,26 +78,30 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
}
- private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
- private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
- private static final LongAdder numTxnOpSuccess = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
- private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+ private final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private final LongAdder numTxnOpSuccess = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
- private static final LongAdder numMessagesAckFailed = new LongAdder();
- private static final LongAdder numMessagesAckSuccess = new LongAdder();
- private static final LongAdder numMessagesSendFailed = new LongAdder();
- private static final LongAdder numMessagesSendSuccess = new LongAdder();
+ private final LongAdder numMessagesAckFailed = new LongAdder();
+ private final LongAdder numMessagesAckSuccess = new LongAdder();
+ private final LongAdder numMessagesSendFailed = new LongAdder();
+ private final LongAdder numMessagesSendSuccess = new LongAdder();
- private static final Recorder messageAckRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static final Recorder messageAckCumulativeRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ // Send and ack latencies are recorded in microseconds. Anything slower than this means the
+ // benchmark is broken rather than slow, so values are clamped to keep HdrHistogram in range.
+ private static final long MAX_LATENCY_MICROS = TimeUnit.HOURS.toMicros(1);
- private static final Recorder messageSendRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static final Recorder messageSendRCumulativeRecorder =
- new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private final Recorder messageAckRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder messageAckCumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+
+ private final Recorder messageSendRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ private final Recorder messageSendRCumulativeRecorder =
+ new Recorder(MAX_LATENCY_MICROS, LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
@Option(names = "--topics-c", description = "All topics that need ack for a transaction", required =
true)
@@ -191,20 +196,6 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
public void run() throws Exception {
super.parseCLI();
- // Reset static counters to avoid stale state from previous runs in the same JVM
- totalNumEndTxnOpFailed.reset();
- totalNumEndTxnOpSuccess.reset();
- numTxnOpSuccess.reset();
- totalNumTxnOpenTxnFail.reset();
- totalNumTxnOpenTxnSuccess.reset();
- numMessagesAckFailed.reset();
- numMessagesAckSuccess.reset();
- numMessagesSendFailed.reset();
- numMessagesSendSuccess.reset();
- messageAckRecorder.reset();
- messageAckCumulativeRecorder.reset();
- messageSendRecorder.reset();
- messageSendRCumulativeRecorder.reset();
// Dump config variables
PerfClientUtils.printJVMInformation(log);
@@ -377,8 +368,8 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
} else {
consumer.acknowledge(message.id());
}
- long latencyMicros = NANOSECONDS.toMicros(
- System.nanoTime() - receiveTime);
+ long latencyMicros = Math.min(NANOSECONDS.toMicros(
+ System.nanoTime() - receiveTime), MAX_LATENCY_MICROS);
messageAckRecorder.recordValue(latencyMicros);
messageAckCumulativeRecorder.recordValue(latencyMicros);
numMessagesAckSuccess.increment();
@@ -413,8 +404,8 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
}
pendingSends.add(msg.send().whenComplete((id, ex) -> {
if (ex == null) {
- long latencyMicros = NANOSECONDS.toMicros(
- System.nanoTime() - sendTime);
+ long latencyMicros = Math.min(NANOSECONDS.toMicros(
+ System.nanoTime() - sendTime), MAX_LATENCY_MICROS);
messageSendRecorder.recordValue(latencyMicros);
messageSendRCumulativeRecorder.recordValue(latencyMicros);
numMessagesSendSuccess.increment();
@@ -584,7 +575,7 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
}
- private static void printTxnAggregatedThroughput(long start) {
+ private void printTxnAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
long numTransactionEndFailed = totalNumEndTxnOpFailed.sum();
long numTransactionEndSuccess = totalNumEndTxnOpSuccess.sum();
@@ -610,7 +601,7 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
}
- private static void printAggregatedThroughput(long start) {
+ private void printAggregatedThroughput(long start) {
double elapsed = (System.nanoTime() - start) / 1e9;
long total = totalNumEndTxnOpFailed.sum() + totalNumEndTxnOpSuccess.sum();
double rate = total / elapsed;
@@ -626,7 +617,7 @@ public class PerformanceTransaction extends PerformanceBaseArguments{
numMessageAckSuccess, numMessageSendSuccess);
}
- private static void printAggregatedStats() {
+ private void printAggregatedStats() {
Histogram reportAckHistogram = messageAckCumulativeRecorder.getIntervalHistogram();
Histogram reportSendHistogram = messageSendRCumulativeRecorder.getIntervalHistogram();
log.infof("Messages ack aggregated latency stats --- Latency: mean: %7.3f ms"
diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java
index 0428ac081bbb..0d1fbd0ceca2 100644
--- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java
+++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerfClientUtilsTest.java
@@ -25,6 +25,8 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import org.HdrHistogram.Histogram;
import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.ProxyProtocol;
@@ -230,6 +232,26 @@ public class PerfClientUtilsTest {
Files.deleteIfExists(testConf);
}
}
+
+ /**
+ * The perf clients hold their latency recorders in static fields, so every subcommand allocates them on
+ * startup rather than only the one being run. HdrHistogram grows the counts array by roughly 10x per
+ * significant digit, and at 5 digits these same ranges cost 11-22 MB each. Pin the bound so raising the
+ * precision again fails here instead of silently costing hundreds of megabytes.
+ */
+ @Test
+ public void latencyHistogramsStaySmallAtTheConfiguredPrecision() {
+ long[] rangesUsedByPerfClients = {
+ TimeUnit.HOURS.toMicros(1), // publish / ack / managed-ledger write latency, in microseconds
+ TimeUnit.DAYS.toMillis(10), // end-to-end consume / read latency, in milliseconds
+ };
+ for (long range : rangesUsedByPerfClients) {
+ Histogram histogram = new Histogram(range, PerfClientUtils.LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS);
+ assertThat(histogram.getEstimatedFootprintInBytes())
+ .as("histogram footprint for range %d", range)
+ .isLessThan(512 * 1024);
+ }
+ }
}
class PerformanceArgumentsTestDefault extends PerformanceBaseArguments {