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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
