congbobo184 commented on a change in pull request #11933:
URL: https://github.com/apache/pulsar/pull/11933#discussion_r715271777
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -304,38 +379,87 @@ public static void main(String[] args) throws Exception {
long latencyMillis = System.currentTimeMillis() -
msg.getPublishTime();
if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
- cumulativeRecorder.recordValue(latencyMillis);
+ recorder.recordValue(latencyMillis);
+ cumulativeRecorder.recordValue(latencyMillis);
+ }
+ if (arguments.isEnableTransaction) {
+ consumer.acknowledgeAsync(msg.getMessageId(),
atomicReference.get()).thenRun(() -> {
+ totalMessageAck.increment();
+ messageAck.increment();
+ }).exceptionally(throwable ->{
+ log.error("Ack message {} failed with exception", msg,
throwable);
+ totalMessageAckFailed.increment();
+ return null;
+ });
+ } else {
+ consumer.acknowledgeAsync(msg).thenRun(()->{
+ totalMessageAck.increment();
+ messageAck.increment();
+ }
+ ).exceptionally(throwable ->{
+ log.error("Ack message {} failed with exception",
msg, throwable);
+ totalMessageAckFailed.increment();
+ return null;
+ }
+ );
}
-
- consumer.acknowledgeAsync(msg);
-
if(arguments.poolMessages) {
msg.release();
}
+ if (arguments.isEnableTransaction
+ && messageAckedCount.incrementAndGet() ==
arguments.numMessagesPerTransaction) {
+ Transaction transaction = atomicReference.get();
+ AtomicBoolean updateTransaction = new AtomicBoolean(true);
+ while (true) {
+
pulsarClient.newTransaction().withTransactionTimeout(arguments.transactionTimeout,
Review comment:
why dont use `buildTransaction`
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -284,14 +320,53 @@ public static void main(String[] args) throws Exception {
final RateLimiter limiter = arguments.rate > 0 ?
RateLimiter.create(arguments.rate) : null;
long startTime = System.nanoTime();
long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+
+ ClientBuilder clientBuilder = PulsarClient.builder() //
+ .enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL) //
+ .connectionsPerBroker(arguments.maxConnections) //
+ .statsInterval(arguments.statsIntervalSeconds,
TimeUnit.SECONDS) //
+ .ioThreads(arguments.ioThreads) //
+ .enableBusyWait(arguments.enableBusyWait)
+ .tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);
+ if (isNotBlank(arguments.authPluginClassName)) {
+ clientBuilder.authentication(arguments.authPluginClassName,
arguments.authParams);
+ }
+
+ if (arguments.tlsAllowInsecureConnection != null) {
+
clientBuilder.allowTlsInsecureConnection(arguments.tlsAllowInsecureConnection);
+ }
+
+ if (isNotBlank(arguments.listenerName)) {
+ clientBuilder.listenerName(arguments.listenerName);
+ }
+ PulsarClient pulsarClient = clientBuilder.build();
+
+ AtomicReference<Transaction> atomicReference =
buildTransaction(pulsarClient, arguments);
+
+ AtomicLong messageAckedCount = new AtomicLong();
+ Semaphore messageReceiveLimiter = new
Semaphore(arguments.numMessagesPerTransaction);
+ Thread thread = Thread.currentThread();
MessageListener<ByteBuffer> listener = (consumer, msg) -> {
+ try {
+ if(arguments.isEnableTransaction){
+ messageReceiveLimiter.acquire();
+ }
Review comment:
format error
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
##########
@@ -680,20 +791,45 @@ private static void runProducer(int producerId,
if (null != client) {
try {
client.close();
+ PerfClientUtils.exit(-1);
} catch (PulsarClientException e) {
log.error("Failed to close test client", e);
}
}
}
}
- private static void printAggregatedThroughput(long start) {
+ private static void printAggregatedThroughput(long start, Arguments
arguments) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesSent.sum() / elapsed;
double throughput = totalBytesSent.sum() / elapsed / 1024 / 1024 * 8;
+ long totalTxnSuccess = 0;
+ long totalTxnFail = 0;
+ double rateOpenTxn = 0;
+ long numTransactionOpenFailed = 0;
+ long numTransactionOpenSuccess = 0;
+ if (arguments.isEnableTransaction) {
+ totalTxnSuccess = totalEndTxnOpSuccessNum.sum();
+ totalTxnFail = totalEndTxnOpFailNum.sum();
+ rateOpenTxn = elapsed / (totalTxnFail + totalTxnSuccess);
+ numTransactionOpenFailed = totalNumTxnOpenTxnFail.sum();
+ numTransactionOpenSuccess = totalNumTxnOpenTxnSuccess.sum();
+ }
+
+ if(arguments.isEnableTransaction){
+ log.info("--- Transaction : {} transaction end successfully --- {}
transaction end failed "
+ + "--- {} transaction open successfully --- {}
transaction open failed "
+ + "--- {} Txn/s",
+ totalTxnSuccess,
+ totalTxnFail,
+ numTransactionOpenSuccess,
+ numTransactionOpenFailed,
+ totalFormat.format(rateOpenTxn));
+ }
log.info(
- "Aggregated throughput stats --- {} records sent --- {} msg/s ---
{} Mbit/s",
- totalMessagesSent,
+ "Aggregated throughput stats --- {} records sent --- {} records
send failed --- {} msg/s --- {} Mbit/s ",
Review comment:
records sent --- {} msg/s
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,717 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOp = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOp = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = 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);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message",
help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need
ack for a transaction", required =
+ true)
+ public List<String> consumerTopic =
Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need
produce for a transaction",
+ required = true)
+ public List<String> producerTopic =
Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description =
"Number of test threads."
+ + "This thread is for new transaction and ack the
consumerTopic message and produce message to "
+ + "producerTopic then commit or abort this transaction. "
+ + "Increasing the number of threads will increase the
parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin
URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar
Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with
the given number of partitions, set 0 "
+ + "to not try to create the topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP
connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If
0, it will keep publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description =
"Set the number of threads to be " +
+ "used for handling connections to brokers, default is 1
thread")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to
consume on (e.g. sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description =
"Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description =
"Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition =
SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description =
"Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description =
"Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the
time value of transaction timeout,"
+ + " and the TimeUnit is second. (Only --txn-enable true can it
take effect) ")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction,
if 0, it will keep opening."
+ + "If transaction disable, it means the number of task. The
task or transaction will produce or "
+ + "consume a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a
transaction."
+ + "If transaction disable, it means the number of
messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a
transaction."
+ + "if transaction disable, it means the number of
message consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or
disable transaction")
+ public boolean isEnableTransaction = true;
+
+ @Parameter(names = {"-commit"}, description = "Whether to commit or
abort the transaction. (Only --txn-enable "
+ + "true can it take effect)")
+ public boolean isCommitTransaction = true;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of
transaction/task open, if 0, it will don`t limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException,
InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain
backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl",
"http://localhost:8080/";);
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL",
"http://localhost:8080/";);
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}",
w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {}
partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic,
arguments.partitions);
+ } catch (PulsarAdminException.ConflictException
alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic,
alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+
client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions !=
arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a
wrong number of partitions: {}, expecting {}",
+ topic,
partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+
PulsarClient.builder().enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService = new
ThreadPoolExecutor(arguments.numTestThreads,
+ arguments.numTestThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>());
+
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (arguments.isEnableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ LongAdder messageSend = new LongAdder();
+ LongAdder messageReceived = new LongAdder();
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in advance,
and then this thread is
+ //responsible for the production and consumption tasks of
the transaction through the loop.
+ //A thread may perform tasks of multiple transactions in a
traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with
exception : ", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ AtomicReference<Transaction> atomicReference =
buildTransaction(client, arguments);
+ //The while loop has no break, and finally ends the
execution through the shutdownNow of
+ //0the executorService
+ while (true) {
+ try {
+ Transaction transaction = atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions :
consumers) {
+ while(true) {
+ if(messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction){
+ break;
+ }
+ for (Consumer<byte[]> consumer :
subscriptions) {
+ if (messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction) {
Review comment:
every sub need ack the number of `numMessagesReceivedPerTransaction `
messages
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
##########
@@ -634,10 +685,15 @@ private static void runProducer(int producerId,
} else {
payloadData = payloadBytes;
}
-
- TypedMessageBuilder<byte[]> messageBuilder =
producer.newMessage()
- .value(payloadData);
- if (arguments.delay >0) {
+ TypedMessageBuilder<byte[]> messageBuilder;
+ if (arguments.isEnableTransaction) {
+ messageBuilder = producer.newMessage(transaction)
Review comment:
this transaction should get under
```
if(arguments.isEnableTransaction &&
arguments.numMessagesPerTransaction > 0){
numMsgPerTxnLimit.acquire();
}
```
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
##########
@@ -454,25 +482,37 @@ public static void main(String[] args) throws Exception {
long now = System.nanoTime();
double elapsed = (now - oldTime) / 1e9;
long total = totalMessagesSent.sum();
+ long totalTxnOpSuccess = 0;
+ long totalTxnOpFail = 0;
+ double rateOpenTxn = 0;
+ if (arguments.isEnableTransaction) {
+ totalTxnOpSuccess = totalEndTxnOpSuccessNum.sum();
+ totalTxnOpFail = totalEndTxnOpFailNum.sum();
+ rateOpenTxn = numTxnOp.sumThenReset() / elapsed;
+ }
double rate = messagesSent.sumThenReset() / elapsed;
- double failureRate = messagesFailed.sumThenReset() / elapsed;
double throughput = bytesSent.sumThenReset() / elapsed / 1024 /
1024 * 8;
reportHistogram = recorder.getIntervalHistogram(reportHistogram);
+ if (arguments.isEnableTransaction) {
+ log.info("--- Transaction : {} transaction end successfully
---{} transaction end failed "
+ + "--- {} Txn/s",
+ totalTxnOpSuccess, totalTxnOpFail,
totalFormat.format(rateOpenTxn));
+ }
log.info(
- "Throughput produced: {} msg --- {} msg/s --- {} Mbit/s
--- failure {} msg/s --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {}
- 99.9pct: {} - 99.99pct: {} - Max: {}",
+ "Throughput produced: {} msg --- {} msg/s --- {} Mbit/s "
+ + "--- Latency: mean: "
+ + "{} ms - med: {} - 95pct: {} - 99pct: {} -
99.9pct: {} - 99.99pct: {} - Max: {}",
intFormat.format(total),
throughputFormat.format(rate),
throughputFormat.format(throughput),
- throughputFormat.format(failureRate),
Review comment:
why delete this?
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
##########
@@ -665,13 +721,68 @@ private static void runProducer(int producerId,
if (ex.getCause() instanceof
ArrayIndexOutOfBoundsException) {
return null;
}
- log.warn("Write error on message", ex);
- messagesFailed.increment();
+ log.warn("Write message error with exception", ex);
+ totalMessagesSendFailed.increment();
if (arguments.exitOnFailure) {
PerfClientUtils.exit(-1);
}
return null;
});
+ if (arguments.isEnableTransaction
+ && numMessageSend.incrementAndGet() ==
arguments.numMessagesPerTransaction) {
+ try {
+ AtomicBoolean updateTransaction = new
AtomicBoolean(true);
+
+ while(true) {
+
pulsarClient.newTransaction().withTransactionTimeout(arguments.transactionTimeout,
Review comment:
use 'buildTransaction', same as above PerformanceConsumer
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -304,38 +379,87 @@ public static void main(String[] args) throws Exception {
long latencyMillis = System.currentTimeMillis() -
msg.getPublishTime();
if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
- cumulativeRecorder.recordValue(latencyMillis);
+ recorder.recordValue(latencyMillis);
+ cumulativeRecorder.recordValue(latencyMillis);
+ }
+ if (arguments.isEnableTransaction) {
+ consumer.acknowledgeAsync(msg.getMessageId(),
atomicReference.get()).thenRun(() -> {
+ totalMessageAck.increment();
+ messageAck.increment();
+ }).exceptionally(throwable ->{
+ log.error("Ack message {} failed with exception", msg,
throwable);
+ totalMessageAckFailed.increment();
+ return null;
+ });
+ } else {
+ consumer.acknowledgeAsync(msg).thenRun(()->{
+ totalMessageAck.increment();
+ messageAck.increment();
+ }
+ ).exceptionally(throwable ->{
+ log.error("Ack message {} failed with exception",
msg, throwable);
+ totalMessageAckFailed.increment();
+ return null;
+ }
+ );
}
-
- consumer.acknowledgeAsync(msg);
-
if(arguments.poolMessages) {
msg.release();
}
+ if (arguments.isEnableTransaction
+ && messageAckedCount.incrementAndGet() ==
arguments.numMessagesPerTransaction) {
+ Transaction transaction = atomicReference.get();
+ AtomicBoolean updateTransaction = new AtomicBoolean(true);
+ while (true) {
+
pulsarClient.newTransaction().withTransactionTimeout(arguments.transactionTimeout,
+
TimeUnit.SECONDS).build().thenAccept(newTransaction -> {
+ atomicReference.compareAndSet(transaction,
newTransaction);
+ totalNumTxnOpenTxnSuccess.increment();
+ if (arguments.isCommitTransaction) {
+ transaction.commit()
+ .thenRun(() -> {
+ totalEndTxnOpSuccessNum.increment();
+ numTxnOp.increment();
+ totalNumTxnOp.increment();
+ })
+ .exceptionally(exception -> {
+ log.error("Commit transaction failed
with exception : ", exception);
+ totalEndTxnOpFailNum.increment();
+ totalNumTxnOp.increment();
+ return null;
+ });
+ } else {
+ transaction.abort().thenRun(() -> {
+ log.info("Abort transaction {}",
transaction.getTxnID().toString());
+ totalEndTxnOpSuccessNum.increment();
+ numTxnOp.increment();
+ totalNumTxnOp.increment();
+ }).exceptionally(exception -> {
+ log.error("Commit transaction {} failed with
exception",
+ transaction.getTxnID().toString(),
+ exception);
+ totalEndTxnOpFailNum.increment();
+ return null;
+ });
+ }
+ messageAckedCount.set(0);
+
messageReceiveLimiter.release(arguments.numMessagesPerTransaction);
+ }).exceptionally(exception -> {
+ log.error("Failed to new transaction with exception:",
exception);
+ updateTransaction.set(false);
+ totalEndTxnOpFailNum.increment();
+ return null;
+ });
+ if(updateTransaction.get()){
Review comment:
This code doesn't seem to work
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -304,38 +379,87 @@ public static void main(String[] args) throws Exception {
long latencyMillis = System.currentTimeMillis() -
msg.getPublishTime();
if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
- cumulativeRecorder.recordValue(latencyMillis);
+ recorder.recordValue(latencyMillis);
+ cumulativeRecorder.recordValue(latencyMillis);
+ }
+ if (arguments.isEnableTransaction) {
+ consumer.acknowledgeAsync(msg.getMessageId(),
atomicReference.get()).thenRun(() -> {
Review comment:
Need to have spaces in the if code block
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -304,38 +379,87 @@ public static void main(String[] args) throws Exception {
long latencyMillis = System.currentTimeMillis() -
msg.getPublishTime();
if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
Review comment:
don't change this code
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,717 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOp = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOp = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = 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);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message",
help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need
ack for a transaction", required =
+ true)
+ public List<String> consumerTopic =
Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need
produce for a transaction",
+ required = true)
+ public List<String> producerTopic =
Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description =
"Number of test threads."
+ + "This thread is for new transaction and ack the
consumerTopic message and produce message to "
+ + "producerTopic then commit or abort this transaction. "
+ + "Increasing the number of threads will increase the
parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin
URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar
Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with
the given number of partitions, set 0 "
+ + "to not try to create the topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP
connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If
0, it will keep publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description =
"Set the number of threads to be " +
+ "used for handling connections to brokers, default is 1
thread")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to
consume on (e.g. sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description =
"Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description =
"Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition =
SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description =
"Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description =
"Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the
time value of transaction timeout,"
+ + " and the TimeUnit is second. (Only --txn-enable true can it
take effect) ")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction,
if 0, it will keep opening."
+ + "If transaction disable, it means the number of task. The
task or transaction will produce or "
+ + "consume a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a
transaction."
+ + "If transaction disable, it means the number of
messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a
transaction."
+ + "if transaction disable, it means the number of
message consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or
disable transaction")
+ public boolean isEnableTransaction = true;
+
+ @Parameter(names = {"-commit"}, description = "Whether to commit or
abort the transaction. (Only --txn-enable "
+ + "true can it take effect)")
+ public boolean isCommitTransaction = true;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of
transaction/task open, if 0, it will don`t limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException,
InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain
backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl",
"http://localhost:8080/";);
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL",
"http://localhost:8080/";);
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}",
w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {}
partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic,
arguments.partitions);
+ } catch (PulsarAdminException.ConflictException
alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic,
alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+
client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions !=
arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a
wrong number of partitions: {}, expecting {}",
+ topic,
partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+
PulsarClient.builder().enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService = new
ThreadPoolExecutor(arguments.numTestThreads,
+ arguments.numTestThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>());
+
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (arguments.isEnableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ LongAdder messageSend = new LongAdder();
+ LongAdder messageReceived = new LongAdder();
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in advance,
and then this thread is
+ //responsible for the production and consumption tasks of
the transaction through the loop.
+ //A thread may perform tasks of multiple transactions in a
traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with
exception : ", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ AtomicReference<Transaction> atomicReference =
buildTransaction(client, arguments);
+ //The while loop has no break, and finally ends the
execution through the shutdownNow of
+ //0the executorService
+ while (true) {
+ try {
+ Transaction transaction = atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions :
consumers) {
+ while(true) {
+ if(messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction){
+ break;
+ }
+ for (Consumer<byte[]> consumer :
subscriptions) {
+ if (messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction) {
+ break;
+ }
+ Message message = null;
+ try {
+ message = consumer.receive();
Review comment:
receive() don't throw exception
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,717 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOp = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOp = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = 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);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message",
help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need
ack for a transaction", required =
+ true)
+ public List<String> consumerTopic =
Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need
produce for a transaction",
+ required = true)
+ public List<String> producerTopic =
Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description =
"Number of test threads."
+ + "This thread is for new transaction and ack the
consumerTopic message and produce message to "
+ + "producerTopic then commit or abort this transaction. "
+ + "Increasing the number of threads will increase the
parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin
URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar
Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with
the given number of partitions, set 0 "
+ + "to not try to create the topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP
connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If
0, it will keep publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description =
"Set the number of threads to be " +
+ "used for handling connections to brokers, default is 1
thread")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to
consume on (e.g. sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description =
"Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description =
"Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition =
SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description =
"Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description =
"Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the
time value of transaction timeout,"
+ + " and the TimeUnit is second. (Only --txn-enable true can it
take effect) ")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction,
if 0, it will keep opening."
+ + "If transaction disable, it means the number of task. The
task or transaction will produce or "
+ + "consume a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a
transaction."
+ + "If transaction disable, it means the number of
messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a
transaction."
+ + "if transaction disable, it means the number of
message consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or
disable transaction")
+ public boolean isEnableTransaction = true;
+
+ @Parameter(names = {"-commit"}, description = "Whether to commit or
abort the transaction. (Only --txn-enable "
+ + "true can it take effect)")
+ public boolean isCommitTransaction = true;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of
transaction/task open, if 0, it will don`t limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException,
InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain
backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl",
"http://localhost:8080/";);
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL",
"http://localhost:8080/";);
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}",
w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {}
partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic,
arguments.partitions);
+ } catch (PulsarAdminException.ConflictException
alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic,
alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+
client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions !=
arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a
wrong number of partitions: {}, expecting {}",
+ topic,
partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+
PulsarClient.builder().enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService = new
ThreadPoolExecutor(arguments.numTestThreads,
+ arguments.numTestThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>());
+
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (arguments.isEnableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ LongAdder messageSend = new LongAdder();
+ LongAdder messageReceived = new LongAdder();
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in advance,
and then this thread is
+ //responsible for the production and consumption tasks of
the transaction through the loop.
+ //A thread may perform tasks of multiple transactions in a
traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with
exception : ", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ AtomicReference<Transaction> atomicReference =
buildTransaction(client, arguments);
+ //The while loop has no break, and finally ends the
execution through the shutdownNow of
+ //0the executorService
+ while (true) {
+ try {
+ Transaction transaction = atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions :
consumers) {
+ while(true) {
+ if(messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction){
+ break;
+ }
+ for (Consumer<byte[]> consumer :
subscriptions) {
+ if (messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction) {
+ break;
+ }
+ Message message = null;
+ try {
+ message = consumer.receive();
+ log.info("Receive message {} ",
message);
+ } catch (Exception e) {
+ log.error("{} can`t receive
message in 2 sec with exception",
+ consumer, e);
+ }
+
+ messageReceived.increment();
+ long receiveTime = System.nanoTime();
+ if (arguments.isEnableTransaction) {
+
consumer.acknowledgeAsync(message.getMessageId(), transaction)
+ .thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+
System.nanoTime() - receiveTime);
+
messageAckRecorder.recordValue(latencyMicros);
+
messageAckCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception
-> {
+ if (exception instanceof
InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message failed
with transaction {} throw exception",
+ transaction,
exception);
+
numMessagesAckFailed.increment();
+ return null;
+ });
+ } else {
+
consumer.acknowledgeAsync(message).thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime() -
receiveTime);
+
messageAckRecorder.recordValue(latencyMicros);
+
messageAckCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if (exception instanceof
InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message failed
with transaction {} throw exception",
+ transaction,
exception);
+
numMessagesAckFailed.increment();
+ return null;
+ });
+ }
+ }
+ }
+ messageReceived.reset();
+ }
+
+ for(Producer<byte[]> producer : producers){
+ while (true){
+ if(messageSend.sum() >=
arguments.numMessagesProducedPerTransaction){
+ break;
+ }
+ long sendTime = System.nanoTime();
+ messageSend.increment();
+ if (arguments.isEnableTransaction) {
+
producer.newMessage(transaction).value(payloadBytes)
+ .sendAsync().thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime() -
sendTime);
+
messageSendRecorder.recordValue(latencyMicros);
+
messageSendRCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if(exception instanceof
InterruptedException && ! executing.get()){
+ return null;
+ }
+ log.error("Send transaction
message failed with exception : ", exception);
+ numMessagesSendFailed.increment();
+ return null;
+ });
+ } else {
+
producer.newMessage().value(payloadBytes)
+ .sendAsync().thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime() -
sendTime);
+
messageSendRecorder.recordValue(latencyMicros);
+
messageSendRCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if(exception instanceof
InterruptedException && ! executing.get()){
+ return null;
+ }
+ log.error("Send message failed
with exception : ", exception);
+ numMessagesSendFailed.increment();
+ return null;
+ });
+ }
+ }
+ messageSend.reset();
+ }
+
+ if(rateLimiter != null){
+ rateLimiter.tryAcquire();
+ }
+ if (arguments.isEnableTransaction) {
+ if (arguments.isCommitTransaction) {
+ log.info("Committing transaction {}",
transaction.getTxnID().toString());
+ transaction.commit()
+ .thenRun(()->{
+
totalNumEndTxnOpSuccess.increment();
+ log.info("Committed
transaction {}", transaction.getTxnID().toString());
+ })
+ .exceptionally(exception -> {
+ if(exception instanceof
InterruptedException && ! executing.get()){
+ return null;
+ }
+ log.error("Commit transaction
{} failed with exception",
+
transaction.getTxnID().toString(),
+ exception);
+
totalNumEndTxnOpFailed.increment();
+ return null;
+ });
+ } else {
+ log.info("Aborting transaction {}",
transaction.getTxnID().toString());
+ transaction.abort().thenRun(() -> {
+ log.info("Abort transaction {}",
transaction.getTxnID().toString());
+ totalNumEndTxnOpSuccess.increment();
+ }).exceptionally(exception -> {
+ if(exception instanceof
InterruptedException && ! executing.get()){
+ return null;
+ }
+ log.error("Commit transaction {}
failed with exception",
+
transaction.getTxnID().toString(),
+ exception);
+ totalNumEndTxnOpFailed.increment();
+ return null;
+ });
+ }
+ AtomicBoolean updateTransaction = new
AtomicBoolean(true);
+
+ while(true) {
+ atomicReference.compareAndSet(transaction,
client.newTransaction()
Review comment:
same as PerformanceProducer and PerformanceConsumer
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,717 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOp = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOp = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = 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);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message",
help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need
ack for a transaction", required =
+ true)
+ public List<String> consumerTopic =
Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need
produce for a transaction",
+ required = true)
+ public List<String> producerTopic =
Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description =
"Number of test threads."
+ + "This thread is for new transaction and ack the
consumerTopic message and produce message to "
+ + "producerTopic then commit or abort this transaction. "
+ + "Increasing the number of threads will increase the
parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin
URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar
Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with
the given number of partitions, set 0 "
+ + "to not try to create the topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP
connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If
0, it will keep publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description =
"Set the number of threads to be " +
+ "used for handling connections to brokers, default is 1
thread")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to
consume on (e.g. sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description =
"Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description =
"Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition =
SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description =
"Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description =
"Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the
time value of transaction timeout,"
+ + " and the TimeUnit is second. (Only --txn-enable true can it
take effect) ")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction,
if 0, it will keep opening."
+ + "If transaction disable, it means the number of task. The
task or transaction will produce or "
+ + "consume a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a
transaction."
+ + "If transaction disable, it means the number of
messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a
transaction."
+ + "if transaction disable, it means the number of
message consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or
disable transaction")
+ public boolean isEnableTransaction = true;
+
+ @Parameter(names = {"-commit"}, description = "Whether to commit or
abort the transaction. (Only --txn-enable "
+ + "true can it take effect)")
+ public boolean isCommitTransaction = true;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of
transaction/task open, if 0, it will don`t limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException,
InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain
backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl",
"http://localhost:8080/";);
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL",
"http://localhost:8080/";);
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}",
w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {}
partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic,
arguments.partitions);
+ } catch (PulsarAdminException.ConflictException
alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic,
alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+
client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions !=
arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a
wrong number of partitions: {}, expecting {}",
+ topic,
partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+
PulsarClient.builder().enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService = new
ThreadPoolExecutor(arguments.numTestThreads,
+ arguments.numTestThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>());
+
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (arguments.isEnableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ LongAdder messageSend = new LongAdder();
+ LongAdder messageReceived = new LongAdder();
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in advance,
and then this thread is
+ //responsible for the production and consumption tasks of
the transaction through the loop.
+ //A thread may perform tasks of multiple transactions in a
traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with
exception : ", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ AtomicReference<Transaction> atomicReference =
buildTransaction(client, arguments);
+ //The while loop has no break, and finally ends the
execution through the shutdownNow of
+ //0the executorService
+ while (true) {
+ try {
+ Transaction transaction = atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions :
consumers) {
+ while(true) {
+ if(messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction){
Review comment:
this code should delete?
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,685 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOp = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOp = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = 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);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message",
help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need
ack for a transaction", required =
+ true)
+ public List<String> consumerTopic =
Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need
produce for a transaction",
+ required = true)
+ public List<String> producerTopic =
Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description =
"Number of test threads."
+ + "This thread is for new transaction and ack the
consumerTopic message and produce message to "
+ + "producerTopic then commit or abort this transaction. "
+ + "Increasing the number of threads will increase the
parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin
URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar
Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with
the given number of partitions, set 0 "
+ + "to not try to create the topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP
connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If
0, it will keep publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description =
"Set the number of threads to be " +
+ "used for handling connections to brokers, default is 1
thread")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to
consume on (e.g. sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description =
"Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description =
"Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition =
SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description =
"Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description =
"Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the
time value of transaction timeout,"
+ + " and the TimeUnit is second. (Only --txn-enable true can it
take effect) ")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction,
if 0, it will keep opening."
+ + "If transaction disable, it means the number of task. The
task or transaction will produce or "
+ + "and consume a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a
transaction."
+ + "If transaction disable, it means the number of
messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a
transaction."
+ + "if transaction disable, it means the number of
message consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or
disable transaction")
+ public boolean isEnableTransaction = true;
+
+ @Parameter(names = {"-commit"}, description = "Whether to commit or
abort the transaction. (Only --txn-enable "
+ + "true can it take effect)")
+ public boolean isCommitTransaction = true;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of
transaction/task open, if 0, it will don`t limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException,
InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain
backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl",
"http://localhost:8080/";);
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL",
"http://localhost:8080/";);
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}",
w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {}
partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic,
arguments.partitions);
+ } catch (PulsarAdminException.ConflictException
alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic,
alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+
client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions !=
arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a
wrong number of partitions: {}, expecting {}",
+ topic,
partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+
PulsarClient.builder().enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService =
Executors.newFixedThreadPool(arguments.numTestThreads);
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (arguments.isEnableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+ new Thread(() -> {
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ LongAdder messageSend = new LongAdder();
+ LongAdder messageReceived = new LongAdder();
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in
advance, and then this thread is
+ //responsible for the production and consumption tasks
of the transaction through the loop.
+ //A thread may perform tasks of multiple transactions
in a traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with
exception : " + e);
+ }
+ AtomicReference<Transaction> atomicReference =
buildTransaction(client, arguments);
+ //The while loop has no break, and finally ends the
execution through the shutdownNow of
+ //0the executorService
+ while (true) {
+ try {
+ Transaction transaction =
atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions :
consumers) {
+ while(true) {
+ if(messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction){
+ break;
+ }
+ for (Consumer<byte[]> consumer :
subscriptions) {
+ if (messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction) {
+ break;
+ }
+ Message message = null;
+ try {
+ message = consumer.receive(2,
TimeUnit.SECONDS);
+ log.info("Receive message {}
", message);
+ } catch (Exception e) {
+ log.error("{} can`t receive
message in 2 sec with exception {}",
+ consumer, e);
+ }
+
+ messageReceived.increment();
+ long receiveTime =
System.nanoTime();
+ if (arguments.isEnableTransaction)
{
+
consumer.acknowledgeAsync(message.getMessageId(), transaction)
+ .thenRun(() -> {
+ long latencyMicros
= NANOSECONDS.toMicros(
+
System.nanoTime() - receiveTime);
+
messageAckRecorder.recordValue(latencyMicros);
+
messageAckCumulativeRecorder.recordValue(latencyMicros);
+
}).exceptionally(exception -> {
+ if (exception instanceof
InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message
failed with transaction {} throw exception {}",
+ transaction,
exception);
+
numMessagesAckFailed.increment();
+ return null;
+ });
+ } else {
+
consumer.acknowledgeAsync(message).thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime()
- receiveTime);
+
messageAckRecorder.recordValue(latencyMicros);
+
messageAckCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if (exception instanceof
InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message
failed with transaction {} throw exception {}",
+ transaction,
exception);
+
numMessagesAckFailed.increment();
+ return null;
+ });
+ }
+ }
+ }
+ messageReceived.reset();
+ }
+
+ for(Producer<byte[]> producer : producers){
+ while (true){
+ if(messageSend.sum() >=
arguments.numMessagesProducedPerTransaction){
Review comment:
use for loop is better
##########
File path:
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,717 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOp = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOp = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = 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);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message",
help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need
ack for a transaction", required =
+ true)
+ public List<String> consumerTopic =
Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need
produce for a transaction",
+ required = true)
+ public List<String> producerTopic =
Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description =
"Number of test threads."
+ + "This thread is for new transaction and ack the
consumerTopic message and produce message to "
+ + "producerTopic then commit or abort this transaction. "
+ + "Increasing the number of threads will increase the
parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin
URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar
Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with
the given number of partitions, set 0 "
+ + "to not try to create the topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP
connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If
0, it will keep publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description =
"Set the number of threads to be " +
+ "used for handling connections to brokers, default is 1
thread")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to
consume on (e.g. sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description =
"Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description =
"Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition =
SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description =
"Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description =
"Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the
time value of transaction timeout,"
+ + " and the TimeUnit is second. (Only --txn-enable true can it
take effect) ")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction,
if 0, it will keep opening."
+ + "If transaction disable, it means the number of task. The
task or transaction will produce or "
+ + "consume a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a
transaction."
+ + "If transaction disable, it means the number of
messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a
transaction."
+ + "if transaction disable, it means the number of
message consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or
disable transaction")
+ public boolean isEnableTransaction = true;
+
+ @Parameter(names = {"-commit"}, description = "Whether to commit or
abort the transaction. (Only --txn-enable "
+ + "true can it take effect)")
+ public boolean isCommitTransaction = true;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of
transaction/task open, if 0, it will don`t limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException,
InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain
backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl",
"http://localhost:8080/";);
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL",
"http://localhost:8080/";);
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}",
w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {}
partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic,
arguments.partitions);
+ } catch (PulsarAdminException.ConflictException
alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic,
alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+
client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions !=
arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a
wrong number of partitions: {}, expecting {}",
+ topic,
partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+
PulsarClient.builder().enableTransaction(arguments.isEnableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService = new
ThreadPoolExecutor(arguments.numTestThreads,
+ arguments.numTestThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>());
+
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (arguments.isEnableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ LongAdder messageSend = new LongAdder();
+ LongAdder messageReceived = new LongAdder();
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in advance,
and then this thread is
+ //responsible for the production and consumption tasks of
the transaction through the loop.
+ //A thread may perform tasks of multiple transactions in a
traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with
exception : ", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ AtomicReference<Transaction> atomicReference =
buildTransaction(client, arguments);
+ //The while loop has no break, and finally ends the
execution through the shutdownNow of
+ //0the executorService
+ while (true) {
+ try {
+ Transaction transaction = atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions :
consumers) {
+ while(true) {
+ if(messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction){
+ break;
+ }
+ for (Consumer<byte[]> consumer :
subscriptions) {
+ if (messageReceived.sum() ==
arguments.numMessagesReceivedPerTransaction) {
+ break;
+ }
+ Message message = null;
+ try {
+ message = consumer.receive();
+ log.info("Receive message {} ",
message);
+ } catch (Exception e) {
+ log.error("{} can`t receive
message in 2 sec with exception",
+ consumer, e);
+ }
+
+ messageReceived.increment();
+ long receiveTime = System.nanoTime();
+ if (arguments.isEnableTransaction) {
+
consumer.acknowledgeAsync(message.getMessageId(), transaction)
+ .thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+
System.nanoTime() - receiveTime);
+
messageAckRecorder.recordValue(latencyMicros);
+
messageAckCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception
-> {
+ if (exception instanceof
InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message failed
with transaction {} throw exception",
+ transaction,
exception);
+
numMessagesAckFailed.increment();
+ return null;
+ });
+ } else {
+
consumer.acknowledgeAsync(message).thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime() -
receiveTime);
+
messageAckRecorder.recordValue(latencyMicros);
+
messageAckCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if (exception instanceof
InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message failed
with transaction {} throw exception",
+ transaction,
exception);
+
numMessagesAckFailed.increment();
+ return null;
+ });
+ }
+ }
+ }
+ messageReceived.reset();
+ }
+
+ for(Producer<byte[]> producer : producers){
+ while (true){
+ if(messageSend.sum() >=
arguments.numMessagesProducedPerTransaction){
+ break;
+ }
+ long sendTime = System.nanoTime();
+ messageSend.increment();
+ if (arguments.isEnableTransaction) {
+
producer.newMessage(transaction).value(payloadBytes)
+ .sendAsync().thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime() -
sendTime);
+
messageSendRecorder.recordValue(latencyMicros);
+
messageSendRCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if(exception instanceof
InterruptedException && ! executing.get()){
+ return null;
+ }
+ log.error("Send transaction
message failed with exception : ", exception);
+ numMessagesSendFailed.increment();
+ return null;
+ });
+ } else {
+
producer.newMessage().value(payloadBytes)
+ .sendAsync().thenRun(() -> {
+ long latencyMicros =
NANOSECONDS.toMicros(
+ System.nanoTime() -
sendTime);
+
messageSendRecorder.recordValue(latencyMicros);
+
messageSendRCumulativeRecorder.recordValue(latencyMicros);
+ }).exceptionally(exception -> {
+ if(exception instanceof
InterruptedException && ! executing.get()){
+ return null;
+ }
+ log.error("Send message failed
with exception : ", exception);
+ numMessagesSendFailed.increment();
+ return null;
+ });
+ }
+ }
+ messageSend.reset();
+ }
+
+ if(rateLimiter != null){
Review comment:
this should process after current transaction commit or abort
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
