You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/10/11 02:57:13 UTC
[GitHub] [pulsar] Anonymitaet commented on a change in pull request #11933: [Transaction] Add transaction perf
Anonymitaet commented on a change in pull request #11933:
URL: https://github.com/apache/pulsar/pull/11933#discussion_r725755283
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
##########
@@ -250,6 +263,22 @@
@Parameter(names = {"-fc", "--format-class"}, description="Custom Formatter class name")
public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
+
+ @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 = 10;
+
+ @Parameter(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages per transaction send. "
Review comment:
```suggestion
description = "The number of messages sent by a transaction. "
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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.")
Review comment:
```suggestion
+ "If transaction disabled, it means the number of messages consumed in a task.")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -182,6 +198,26 @@
@Parameter(names = {"-bw", "--busy-wait"}, description = "Enable Busy-Wait on the Pulsar client")
public boolean enableBusyWait = false;
+
+ @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 = {"-nmt", "---numMessage-perTransaction"},
+ description = "The number of messages per transaction acknowledgment. "
+ + "(Only --txn-enable true can it take effect")
+ public int numMessagesPerTransaction = 50;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
+
+ @Parameter(names = {"-ntxn"}, description = "The number of transaction will be opened, if 0, it will keep open."
Review comment:
```suggestion
@Parameter(names = {"-ntxn"}, description = "The number of opened transactions. 0 means keeping open."
```
Do you mean this?
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
##########
@@ -182,6 +198,26 @@
@Parameter(names = {"-bw", "--busy-wait"}, description = "Enable Busy-Wait on the Pulsar client")
public boolean enableBusyWait = false;
+
+ @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)")
Review comment:
```suggestion
+ " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
```
Please check all occurrences
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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")
Review comment:
```suggestion
"--partitions"}, description = "Create partitioned topics with a given number of partitions. 0 means"
+ "not trying to create a topic")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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")
Review comment:
```suggestion
"--test-duration"}, description = "Test duration (in second). 0 means keeping publishing")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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)")
Review comment:
```suggestion
"--subscriptions"}, description = "A list of subscriptions to consume (for example, sub1,sub2)")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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")
Review comment:
```suggestion
"used for handling connections to brokers. The default value is 1.")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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.")
Review comment:
```suggestion
+ "If transaction disabled, it means the number of messages produced in a task.")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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")
Review comment:
```suggestion
@Parameter(names = "-txnRate", description = "Set the rate of opened transaction or task. 0 means no limit")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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.")
Review comment:
```suggestion
+ "This thread is for a new transaction to ack the messages from consumer topics and produce message to "
+ "producer topics, and then commit or abort this transaction. "
+ "Increasing the number of threads increases the parallelism of the performance test, "
+ "thereby increasing the intensity of the stress test.")
```
Do you mean this?
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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.")
Review comment:
```suggestion
"--number-txn"}, description = "Set the number of transaction. 0 means keeping open."
+ "If transaction disabled, it means the number of tasks. The task or transaction produces or "
+ "consumes a specified number of messages.")
```
##########
File path: pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
##########
@@ -0,0 +1,697 @@
+/**
+ * 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 static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+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.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.AtomicLong;
+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.PulsarClientException;
+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 totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private 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."
Review comment:
```suggestion
description = "Set the number of messages consumed in a transaction."
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org