You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "crossoverJie (via GitHub)" <gi...@apache.org> on 2024/03/31 08:04:17 UTC
[PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
crossoverJie opened a new pull request, #22388:
URL: https://github.com/apache/pulsar/pull/22388
PIP: #22181
Improve CLI: `pulsar-perf` user experience.
### Motivation
Refactor `pulsar-perf` to subcommand.
### Modifications
- Create a new class `PulsarPerfTestTool`
- Remove unnecessary code in `pulsar-perf` and `pulsar-perf.cmd`
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: https://github.com/crossoverJie/pulsar/pull/22
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566012999
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java:
##########
@@ -713,20 +688,22 @@ public void run() throws Exception {
/**
* Start a controller with command line arguments.
*
- * @param args
- * Arguments to pass in.
*/
- public static void main(String[] args) throws Exception {
- final MainArguments arguments = new MainArguments();
- final CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf simulation-controller");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1568319548
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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 org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import java.io.FileInputStream;
+import java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
+ showDefaultValues = true
+)
+public class PulsarPerfTestTool {
+
+ protected Map<String, Class<?>> commandMap;
+ protected final CommandLine commander;
+
+ public PulsarPerfTestTool() {
+ this.commander = new CommandLine(this);
+ commandMap = new HashMap<>();
+ }
+
+ private String[] initCommander(String[] args) throws Exception {
+ commandMap.put("produce", PerformanceProducer.class);
+ commandMap.put("consume", PerformanceConsumer.class);
+ commandMap.put("transaction", PerformanceTransaction.class);
+ commandMap.put("read", PerformanceReader.class);
+ commandMap.put("monitor-brokers", BrokerMonitor.class);
+ commandMap.put("simulation-client", LoadSimulationClient.class);
+ commandMap.put("simulation-controller", LoadSimulationController.class);
+ commandMap.put("websocket-producer", PerformanceClient.class);
+ commandMap.put("managed-ledger", ManagedLedgerWriter.class);
+ commandMap.put("gen-doc", CmdGenerateDocumentation.class);
+ if (args.length == 0) {
+ System.out.println("Usage: pulsar-perf CONF_FILE_PATH [options] [command] [command options]");
+ PerfClientUtils.exit(0);
+ }
+ String configFile = args[0];
+ Properties prop = new Properties(System.getProperties());
+ if (configFile != null) {
+ try (FileInputStream fis = new FileInputStream(configFile)) {
+ prop.load(fis);
+ }
+ }
+
+ for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
+ if (PerformanceBaseArguments.class.isAssignableFrom(c.getValue())){
+ Constructor<?> constructor = c.getValue().getDeclaredConstructor();
+ constructor.setAccessible(true);
+ commander.setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
Review Comment:
You are right, done.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java:
##########
@@ -79,86 +78,69 @@ public class ManagedLedgerWriter {
private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
- @Command(description = "Write directly on managed-ledgers", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments {
- @Option(names = { "-h", "--help" }, description = "Help message", help = true)
- boolean help;
+ @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
+ public int msgRate = 100;
- @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
- public int msgRate = 100;
+ @Option(names = { "-s", "--size" }, description = "Message size")
+ public int msgSize = 1024;
- @Option(names = { "-s", "--size" }, description = "Message size")
- public int msgSize = 1024;
+ @Option(names = { "-t", "--num-topic" },
+ description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
+ public int numManagedLedgers = 1;
- @Option(names = { "-t", "--num-topic" },
- description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
- public int numManagedLedgers = 1;
+ @Option(names = { "--threads" },
+ description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
+ public int numThreads = 1;
- @Option(names = { "--threads" },
- description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
- public int numThreads = 1;
+ @Deprecated
+ @Option(names = {"-zk", "--zookeeperServers"},
+ description = "ZooKeeper connection string",
+ hidden = true)
+ public String zookeeperServers;
- @Deprecated
- @Option(names = {"-zk", "--zookeeperServers"},
- description = "ZooKeeper connection string",
- hidden = true)
- public String zookeeperServers;
+ @Option(names = {"-md",
+ "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
+ private String metadataStoreUrl;
- @Option(names = {"-md",
- "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
- private String metadataStoreUrl;
+ @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
+ public int maxOutstanding = 1000;
- @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
- public int maxOutstanding = 1000;
+ @Option(names = { "-c",
+ "--max-connections" }, description = "Max number of TCP connections to a single bookie")
+ public int maxConnections = 1;
- @Option(names = { "-c",
- "--max-connections" }, description = "Max number of TCP connections to a single bookie")
- public int maxConnections = 1;
+ @Option(names = { "-m",
+ "--num-messages" },
+ description = "Number of messages to publish in total. If <= 0, it will keep publishing")
+ public long numMessages = 0;
- @Option(names = { "-m",
- "--num-messages" },
- description = "Number of messages to publish in total. If <= 0, it will keep publishing")
- public long numMessages = 0;
+ @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
+ public int ensembleSize = 1;
- @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
- public int ensembleSize = 1;
+ @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
+ public int writeQuorum = 1;
- @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
- public int writeQuorum = 1;
+ @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
+ public int ackQuorum = 1;
- @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
- public int ackQuorum = 1;
+ @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
+ public DigestType digestType = DigestType.CRC32C;
- @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
- public DigestType digestType = DigestType.CRC32C;
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
+ public ManagedLedgerWriter() {
+ super("managed-ledger");
}
- public static void main(String[] args) throws Exception {
-
- final Arguments arguments = new Arguments();
- CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf managed-ledger");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
-
- if (arguments.help) {
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
Done.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1559622685
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java:
##########
@@ -24,57 +24,46 @@
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.ScopeType;
@Slf4j
-public class CmdGenerateDocumentation {
+@Command(description = "Generate documentation automatically.", showDefaultValues = true, scope = ScopeType.INHERIT)
+public class CmdGenerateDocumentation extends CmdBase{
- @Command(description = "Generate documentation automatically.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments {
- @Option(names = {"-h", "--help"}, description = "Help message", help = true)
- boolean help;
+ @Option(names = {"-h", "--help"}, description = "Help message", help = true)
+ boolean help;
- @Option(names = {"-n", "--command-names"}, description = "List of command names")
- private List<String> commandNames = new ArrayList<>();
+ @Option(names = {"-n", "--command-names"}, description = "List of command names")
+ private List<String> commandNames = new ArrayList<>();
+ public CmdGenerateDocumentation() {
+ super("gen-doc");
}
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf gen-doc");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
-
- if (arguments.help) {
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
+ if (this.help) {
commander.usage(commander.getOut());
PerfClientUtils.exit(1);
}
Map<String, Class<?>> cmdClassMap = new LinkedHashMap<>();
- cmdClassMap.put("produce", Class.forName("org.apache.pulsar.testclient.PerformanceProducer$Arguments"));
- cmdClassMap.put("consume", Class.forName("org.apache.pulsar.testclient.PerformanceConsumer$Arguments"));
- cmdClassMap.put("transaction", Class.forName("org.apache.pulsar.testclient.PerformanceTransaction$Arguments"));
- cmdClassMap.put("read", Class.forName("org.apache.pulsar.testclient.PerformanceReader$Arguments"));
- cmdClassMap.put("monitor-brokers", Class.forName("org.apache.pulsar.testclient.BrokerMonitor$Arguments"));
- cmdClassMap.put("simulation-client",
- Class.forName("org.apache.pulsar.testclient.LoadSimulationClient$MainArguments"));
- cmdClassMap.put("simulation-controller",
- Class.forName("org.apache.pulsar.testclient.LoadSimulationController$MainArguments"));
- cmdClassMap.put("websocket-producer",
- Class.forName("org.apache.pulsar.proxy.socket.client.PerformanceClient$Arguments"));
- cmdClassMap.put("managed-ledger", Class.forName("org.apache.pulsar.testclient.ManagedLedgerWriter$Arguments"));
+ cmdClassMap.put("produce", PerformanceProducer.class);
Review Comment:
https://github.com/apache/pulsar/blob/434ec1b884b26af6c3658df3a7b90c432e464973/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdGenerateDocument.java#L43-L46
You are right; I found another one.
But they are in different modules, and I think it's a bit difficult to merge them into one.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1559614132
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
Review Comment:
Done.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
+ showDefaultValues = true
+)
+public class PulsarPerfTestTool {
+
+ protected Map<String, Class<?>> commandMap;
+ protected final CommandLine commander;
+
+ public PulsarPerfTestTool() {
+ this.commander = new CommandLine(this);
+ commandMap = new HashMap<>();
+ }
+
+ private String[] initCommander(String[] args) throws Exception {
+ commandMap.put("produce", PerformanceProducer.class);
+ commandMap.put("consume", PerformanceConsumer.class);
+ commandMap.put("transaction", PerformanceTransaction.class);
+ commandMap.put("read", PerformanceReader.class);
+ commandMap.put("monitor-brokers", BrokerMonitor.class);
+ commandMap.put("simulation-client", LoadSimulationClient.class);
+ commandMap.put("simulation-controller", LoadSimulationController.class);
+ commandMap.put("websocket-producer", PerformanceClient.class);
+ commandMap.put("managed-ledger", ManagedLedgerWriter.class);
+ commandMap.put("gen-doc", CmdGenerateDocumentation.class);
+ if (args.length == 0) {
+ System.out.println("Usage: pulsar-perf CONF_FILE_PATH [options] [command] [command options]");
+ PerfClientUtils.exit(0);
+ }
+ String configFile = args[0];
+
+ for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
Review Comment:
Done.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece merged PR #22388:
URL: https://github.com/apache/pulsar/pull/22388
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566043648
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java:
##########
@@ -79,86 +78,69 @@ public class ManagedLedgerWriter {
private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
- @Command(description = "Write directly on managed-ledgers", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments {
- @Option(names = { "-h", "--help" }, description = "Help message", help = true)
- boolean help;
+ @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
+ public int msgRate = 100;
- @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
- public int msgRate = 100;
+ @Option(names = { "-s", "--size" }, description = "Message size")
+ public int msgSize = 1024;
- @Option(names = { "-s", "--size" }, description = "Message size")
- public int msgSize = 1024;
+ @Option(names = { "-t", "--num-topic" },
+ description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
+ public int numManagedLedgers = 1;
- @Option(names = { "-t", "--num-topic" },
- description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
- public int numManagedLedgers = 1;
+ @Option(names = { "--threads" },
+ description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
+ public int numThreads = 1;
- @Option(names = { "--threads" },
- description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
- public int numThreads = 1;
+ @Deprecated
+ @Option(names = {"-zk", "--zookeeperServers"},
+ description = "ZooKeeper connection string",
+ hidden = true)
+ public String zookeeperServers;
- @Deprecated
- @Option(names = {"-zk", "--zookeeperServers"},
- description = "ZooKeeper connection string",
- hidden = true)
- public String zookeeperServers;
+ @Option(names = {"-md",
+ "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
+ private String metadataStoreUrl;
- @Option(names = {"-md",
- "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
- private String metadataStoreUrl;
+ @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
+ public int maxOutstanding = 1000;
- @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
- public int maxOutstanding = 1000;
+ @Option(names = { "-c",
+ "--max-connections" }, description = "Max number of TCP connections to a single bookie")
+ public int maxConnections = 1;
- @Option(names = { "-c",
- "--max-connections" }, description = "Max number of TCP connections to a single bookie")
- public int maxConnections = 1;
+ @Option(names = { "-m",
+ "--num-messages" },
+ description = "Number of messages to publish in total. If <= 0, it will keep publishing")
+ public long numMessages = 0;
- @Option(names = { "-m",
- "--num-messages" },
- description = "Number of messages to publish in total. If <= 0, it will keep publishing")
- public long numMessages = 0;
+ @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
+ public int ensembleSize = 1;
- @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
- public int ensembleSize = 1;
+ @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
+ public int writeQuorum = 1;
- @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
- public int writeQuorum = 1;
+ @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
+ public int ackQuorum = 1;
- @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
- public int ackQuorum = 1;
+ @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
+ public DigestType digestType = DigestType.CRC32C;
- @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
- public DigestType digestType = DigestType.CRC32C;
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
+ public ManagedLedgerWriter() {
+ super("managed-ledger");
}
- public static void main(String[] args) throws Exception {
-
- final Arguments arguments = new Arguments();
- CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf managed-ledger");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
-
- if (arguments.help) {
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
Please check all code in the PR.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566064279
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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 org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import java.io.FileInputStream;
+import java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
+ showDefaultValues = true
+)
+public class PulsarPerfTestTool {
+
+ protected Map<String, Class<?>> commandMap;
+ protected final CommandLine commander;
+
+ public PulsarPerfTestTool() {
+ this.commander = new CommandLine(this);
+ commandMap = new HashMap<>();
+ }
+
+ private String[] initCommander(String[] args) throws Exception {
+ commandMap.put("produce", PerformanceProducer.class);
+ commandMap.put("consume", PerformanceConsumer.class);
+ commandMap.put("transaction", PerformanceTransaction.class);
+ commandMap.put("read", PerformanceReader.class);
+ commandMap.put("monitor-brokers", BrokerMonitor.class);
+ commandMap.put("simulation-client", LoadSimulationClient.class);
+ commandMap.put("simulation-controller", LoadSimulationController.class);
+ commandMap.put("websocket-producer", PerformanceClient.class);
+ commandMap.put("managed-ledger", ManagedLedgerWriter.class);
+ commandMap.put("gen-doc", CmdGenerateDocumentation.class);
+ if (args.length == 0) {
+ System.out.println("Usage: pulsar-perf CONF_FILE_PATH [options] [command] [command options]");
+ PerfClientUtils.exit(0);
+ }
+ String configFile = args[0];
+ Properties prop = new Properties(System.getProperties());
+ if (configFile != null) {
+ try (FileInputStream fis = new FileInputStream(configFile)) {
+ prop.load(fis);
+ }
+ }
+
+ for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
+ if (PerformanceBaseArguments.class.isAssignableFrom(c.getValue())){
+ Constructor<?> constructor = c.getValue().getDeclaredConstructor();
+ constructor.setAccessible(true);
+ commander.setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
Review Comment:
You can move this line to line 69, and simply this logic.
It looks like you can directly call `addCommand()`.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566016191
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java:
##########
@@ -79,86 +78,69 @@ public class ManagedLedgerWriter {
private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
- @Command(description = "Write directly on managed-ledgers", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments {
- @Option(names = { "-h", "--help" }, description = "Help message", help = true)
- boolean help;
+ @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
+ public int msgRate = 100;
- @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
- public int msgRate = 100;
+ @Option(names = { "-s", "--size" }, description = "Message size")
+ public int msgSize = 1024;
- @Option(names = { "-s", "--size" }, description = "Message size")
- public int msgSize = 1024;
+ @Option(names = { "-t", "--num-topic" },
+ description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
+ public int numManagedLedgers = 1;
- @Option(names = { "-t", "--num-topic" },
- description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
- public int numManagedLedgers = 1;
+ @Option(names = { "--threads" },
+ description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
+ public int numThreads = 1;
- @Option(names = { "--threads" },
- description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
- public int numThreads = 1;
+ @Deprecated
+ @Option(names = {"-zk", "--zookeeperServers"},
+ description = "ZooKeeper connection string",
+ hidden = true)
+ public String zookeeperServers;
- @Deprecated
- @Option(names = {"-zk", "--zookeeperServers"},
- description = "ZooKeeper connection string",
- hidden = true)
- public String zookeeperServers;
+ @Option(names = {"-md",
+ "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
+ private String metadataStoreUrl;
- @Option(names = {"-md",
- "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
- private String metadataStoreUrl;
+ @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
+ public int maxOutstanding = 1000;
- @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
- public int maxOutstanding = 1000;
+ @Option(names = { "-c",
+ "--max-connections" }, description = "Max number of TCP connections to a single bookie")
+ public int maxConnections = 1;
- @Option(names = { "-c",
- "--max-connections" }, description = "Max number of TCP connections to a single bookie")
- public int maxConnections = 1;
+ @Option(names = { "-m",
+ "--num-messages" },
+ description = "Number of messages to publish in total. If <= 0, it will keep publishing")
+ public long numMessages = 0;
- @Option(names = { "-m",
- "--num-messages" },
- description = "Number of messages to publish in total. If <= 0, it will keep publishing")
- public long numMessages = 0;
+ @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
+ public int ensembleSize = 1;
- @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
- public int ensembleSize = 1;
+ @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
+ public int writeQuorum = 1;
- @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
- public int writeQuorum = 1;
+ @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
+ public int ackQuorum = 1;
- @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
- public int ackQuorum = 1;
+ @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
+ public DigestType digestType = DigestType.CRC32C;
- @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
- public DigestType digestType = DigestType.CRC32C;
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
+ public ManagedLedgerWriter() {
+ super("managed-ledger");
}
- public static void main(String[] args) throws Exception {
-
- final Arguments arguments = new Arguments();
- CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf managed-ledger");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
-
- if (arguments.help) {
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
Please use the `@Spec`.
This method returns the root commander, not the current commander.
@Spec
CommandSpec spec;
@Override
public void run() throws Exception {
CommandLine commander = spec.commandLine();
}
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566007716
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,153 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(name = "websocket-producer", description = "Test pulsar websocket producer performance.")
+public class PerformanceClient extends CmdBase {
private static final LongAdder messagesSent = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
private static IMessageFormatter messageFormatter = null;
- private CommandLine commander;
- @Command(description = "Test pulsar websocket producer performance.")
- static class Arguments {
+ @Option(names = { "-cf", "--conf-file" }, description = "Configuration file")
+ public String confFile;
- @Option(names = { "-h", "--help" }, description = "Help message", help = true)
- boolean help;
+ @Option(names = { "-u", "--proxy-url" }, description = "Pulsar Proxy URL, e.g., \"ws://localhost:8080/\"")
+ public String proxyURL;
- @Option(names = { "-cf", "--conf-file" }, description = "Configuration file")
- public String confFile;
+ @Parameters(description = "persistent://tenant/ns/my-topic", arity = "1")
+ public List<String> topics;
- @Option(names = { "-u", "--proxy-url" }, description = "Pulsar Proxy URL, e.g., \"ws://localhost:8080/\"")
- public String proxyURL;
+ @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
+ public int msgRate = 100;
- @Parameters(description = "persistent://tenant/ns/my-topic", arity = "1")
- public List<String> topics;
+ @Option(names = { "-s", "--size" }, description = "Message size in byte")
+ public int msgSize = 1024;
- @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
- public int msgRate = 100;
+ @Option(names = { "-t", "--num-topic" }, description = "Number of topics",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numTopics = 1;
- @Option(names = { "-s", "--size" }, description = "Message size in byte")
- public int msgSize = 1024;
+ @Option(names = { "--auth_plugin" }, description = "Authentication plugin class name", hidden = true)
+ public String deprecatedAuthPluginClassName;
- @Option(names = { "-t", "--num-topic" }, description = "Number of topics",
- converter = PositiveNumberParameterConvert.class
- )
- public int numTopics = 1;
+ @Option(names = { "--auth-plugin" }, description = "Authentication plugin class name")
+ public String authPluginClassName;
- @Option(names = { "--auth_plugin" }, description = "Authentication plugin class name", hidden = true)
- public String deprecatedAuthPluginClassName;
+ @Option(
+ names = { "--auth-params" },
+ description = "Authentication parameters, whose format is determined by the implementation "
+ + "of method `configure` in authentication plugin class, for example \"key1:val1,key2:val2\" "
+ + "or \"{\"key1\":\"val1\",\"key2\":\"val2\"}\".")
+ public String authParams;
- @Option(names = { "--auth-plugin" }, description = "Authentication plugin class name")
- public String authPluginClassName;
+ @Option(names = { "-m",
+ "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep"
+ + " publishing")
+ public long numMessages = 0;
- @Option(
- names = { "--auth-params" },
- description = "Authentication parameters, whose format is determined by the implementation "
- + "of method `configure` in authentication plugin class, for example \"key1:val1,key2:val2\" "
- + "or \"{\"key1\":\"val1\",\"key2\":\"val2\"}\".")
- public String authParams;
+ @Option(names = { "-f", "--payload-file" }, description = "Use payload from a file instead of empty buffer")
+ public String payloadFilename = null;
- @Option(names = { "-m",
- "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep"
- + " publishing")
- public long numMessages = 0;
+ @Option(names = { "-e", "--payload-delimiter" },
+ description = "The delimiter used to split lines when using payload from a file")
+ // here escaping \n since default value will be printed with the help text
+ public String payloadDelimiter = "\\n";
- @Option(names = { "-f", "--payload-file" }, description = "Use payload from a file instead of empty buffer")
- public String payloadFilename = null;
+ @Option(names = { "-fp", "--format-payload" },
+ description = "Format %i as a message index in the stream from producer and/or %t as the timestamp"
+ + " nanoseconds")
+ public boolean formatPayload = false;
- @Option(names = { "-e", "--payload-delimiter" },
- description = "The delimiter used to split lines when using payload from a file")
- // here escaping \n since default value will be printed with the help text
- public String payloadDelimiter = "\\n";
+ @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
+ public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
- @Option(names = { "-fp", "--format-payload" },
- description = "Format %i as a message index in the stream from producer and/or %t as the timestamp"
- + " nanoseconds")
- public boolean formatPayload = false;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
- @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
- public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ public PerformanceClient() {
+ super("websocket-producer");
}
- public Arguments loadArguments(String[] args) {
- Arguments arguments = new Arguments();
- commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf websocket-producer");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
- if (arguments.help) {
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ public void loadArguments() {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
@Spec
private CommandSpec commandSpec;
public void loadArguments() {
CommandLine commander = commandSpec.commandLine();
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1565999431
##########
pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java:
##########
@@ -196,33 +215,39 @@ public void fillArgumentsFromProperties(Properties prop) {
@DataProvider(name = "invalidMemoryLimitCliArgumentProvider")
public Object[][] invalidMemoryLimitCliArgumentProvider() {
- return new Object[][] {
- { new String[]{"-ml","-1"}},
- { new String[]{"-ml","1C"}},
- { new String[]{"--memory-limit", "1Q"}}
+ return new Object[][]{
+ {new String[]{"-ml", "-1"}},
+ {new String[]{"-ml", "1C"}},
+ {new String[]{"--memory-limit", "1Q"}}
};
}
@Test
- public void testMemoryLimitCliArgumentDefault() {
+ public void testMemoryLimitCliArgumentDefault() throws Exception {
for (String cmd : List.of(
"pulsar-perf read",
"pulsar-perf produce",
"pulsar-perf consume",
"pulsar-perf transaction"
)) {
// Arrange
- AtomicBoolean called = new AtomicBoolean();
- final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments() {
+ final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments("") {
@Override
- public void fillArgumentsFromProperties(Properties prop) {
- called.set(true);
+ public void run() throws Exception {
+
}
+
};
- baseArgument.confFile = "./src/test/resources/perf_client1.conf";
+ String confFile = "./src/test/resources/perf_client1.conf";
+ Properties prop = new Properties(System.getProperties());
+ try (FileInputStream fis = new FileInputStream(confFile)) {
+ prop.load(fis);
+ }
+ baseArgument.getCommander().setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+ baseArgument.parse(new String[]{});
// Act
- baseArgument.parseCLI(cmd, new String[]{});
+ baseArgument.parseCLI();
Review Comment:
```suggestion
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#issuecomment-2028596287
@nodece PTAL.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1553360355
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
+ showDefaultValues = true
+)
+public class PulsarPerfTestTool {
+
+ protected Map<String, Class<?>> commandMap;
+ protected final CommandLine commander;
+
+ public PulsarPerfTestTool() {
+ this.commander = new CommandLine(this);
+ commandMap = new HashMap<>();
+ }
+
+ private String[] initCommander(String[] args) throws Exception {
+ commandMap.put("produce", PerformanceProducer.class);
+ commandMap.put("consume", PerformanceConsumer.class);
+ commandMap.put("transaction", PerformanceTransaction.class);
+ commandMap.put("read", PerformanceReader.class);
+ commandMap.put("monitor-brokers", BrokerMonitor.class);
+ commandMap.put("simulation-client", LoadSimulationClient.class);
+ commandMap.put("simulation-controller", LoadSimulationController.class);
+ commandMap.put("websocket-producer", PerformanceClient.class);
+ commandMap.put("managed-ledger", ManagedLedgerWriter.class);
+ commandMap.put("gen-doc", CmdGenerateDocumentation.class);
+ if (args.length == 0) {
+ System.out.println("Usage: pulsar-perf CONF_FILE_PATH [options] [command] [command options]");
+ PerfClientUtils.exit(0);
+ }
+ String configFile = args[0];
+
+ for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
Review Comment:
I know we are using this way in the pulsar-admin, but this is not recommended when using the picocli.
Picocli provides a value provider, you can read the config file to the value provider, and `@Option` with `descriptionKey` to read the config value, please see https://picocli.info/#_propertiesdefaultprovider.
```
commander = new CommandLine(this);
commander = ;
commander.setDefaultValueProvider(new PropertiesDefaultProvider(new File("path/to/config/file.properties")));
@Option(names = {"--auth-plugin"}, descriptionKey = "authPlugin")
String authPluginClassName = null;
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
Review Comment:
When `mixinStandardHelpOption=true`, you can remove `-h, --help` option.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java:
##########
@@ -24,57 +24,46 @@
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.ScopeType;
@Slf4j
-public class CmdGenerateDocumentation {
+@Command(description = "Generate documentation automatically.", showDefaultValues = true, scope = ScopeType.INHERIT)
+public class CmdGenerateDocumentation extends CmdBase{
- @Command(description = "Generate documentation automatically.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments {
- @Option(names = {"-h", "--help"}, description = "Help message", help = true)
- boolean help;
+ @Option(names = {"-h", "--help"}, description = "Help message", help = true)
+ boolean help;
- @Option(names = {"-n", "--command-names"}, description = "List of command names")
- private List<String> commandNames = new ArrayList<>();
+ @Option(names = {"-n", "--command-names"}, description = "List of command names")
+ private List<String> commandNames = new ArrayList<>();
+ public CmdGenerateDocumentation() {
+ super("gen-doc");
}
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf gen-doc");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
-
- if (arguments.help) {
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
+ if (this.help) {
commander.usage(commander.getOut());
PerfClientUtils.exit(1);
}
Map<String, Class<?>> cmdClassMap = new LinkedHashMap<>();
- cmdClassMap.put("produce", Class.forName("org.apache.pulsar.testclient.PerformanceProducer$Arguments"));
- cmdClassMap.put("consume", Class.forName("org.apache.pulsar.testclient.PerformanceConsumer$Arguments"));
- cmdClassMap.put("transaction", Class.forName("org.apache.pulsar.testclient.PerformanceTransaction$Arguments"));
- cmdClassMap.put("read", Class.forName("org.apache.pulsar.testclient.PerformanceReader$Arguments"));
- cmdClassMap.put("monitor-brokers", Class.forName("org.apache.pulsar.testclient.BrokerMonitor$Arguments"));
- cmdClassMap.put("simulation-client",
- Class.forName("org.apache.pulsar.testclient.LoadSimulationClient$MainArguments"));
- cmdClassMap.put("simulation-controller",
- Class.forName("org.apache.pulsar.testclient.LoadSimulationController$MainArguments"));
- cmdClassMap.put("websocket-producer",
- Class.forName("org.apache.pulsar.proxy.socket.client.PerformanceClient$Arguments"));
- cmdClassMap.put("managed-ledger", Class.forName("org.apache.pulsar.testclient.ManagedLedgerWriter$Arguments"));
+ cmdClassMap.put("produce", PerformanceProducer.class);
Review Comment:
I remember we have multiple gen-doc class, you can improve here.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
Review Comment:
```suggestion
@Command(name=""websocket-producer", description = "Test pulsar websocket producer performance.")
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
+public class PerformanceClient extends CmdBase {
private static final LongAdder messagesSent = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
private static IMessageFormatter messageFormatter = null;
- private CommandLine commander;
- @Command(description = "Test pulsar websocket producer performance.")
- static class Arguments {
+ @Option(names = { "-h", "--help" }, description = "Help message", help = true)
+ boolean help;
Review Comment:
Unnecessary.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
+public class PerformanceClient extends CmdBase {
private static final LongAdder messagesSent = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
private static IMessageFormatter messageFormatter = null;
- private CommandLine commander;
- @Command(description = "Test pulsar websocket producer performance.")
- static class Arguments {
+ @Option(names = { "-h", "--help" }, description = "Help message", help = true)
+ boolean help;
Review Comment:
The `mixinStandardHelpOptions = true` has been set in the root command.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
+ showDefaultValues = true
+)
+public class PulsarPerfTestTool {
+
+ protected Map<String, Class<?>> commandMap;
+ protected final CommandLine commander;
+
+ public PulsarPerfTestTool() {
+ this.commander = new CommandLine(this);
+ commandMap = new HashMap<>();
+ }
+
+ private String[] initCommander(String[] args) throws Exception {
+ commandMap.put("produce", PerformanceProducer.class);
+ commandMap.put("consume", PerformanceConsumer.class);
+ commandMap.put("transaction", PerformanceTransaction.class);
+ commandMap.put("read", PerformanceReader.class);
+ commandMap.put("monitor-brokers", BrokerMonitor.class);
+ commandMap.put("simulation-client", LoadSimulationClient.class);
+ commandMap.put("simulation-controller", LoadSimulationController.class);
+ commandMap.put("websocket-producer", PerformanceClient.class);
+ commandMap.put("managed-ledger", ManagedLedgerWriter.class);
+ commandMap.put("gen-doc", CmdGenerateDocumentation.class);
+ if (args.length == 0) {
+ System.out.println("Usage: pulsar-perf CONF_FILE_PATH [options] [command] [command options]");
+ PerfClientUtils.exit(0);
+ }
+ String configFile = args[0];
+
+ for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
Review Comment:
And then you can remove the CmdBase.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
+public class PerformanceClient extends CmdBase {
private static final LongAdder messagesSent = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
private static IMessageFormatter messageFormatter = null;
- private CommandLine commander;
- @Command(description = "Test pulsar websocket producer performance.")
- static class Arguments {
+ @Option(names = { "-h", "--help" }, description = "Help message", help = true)
+ boolean help;
Review Comment:
Other commands also have this issue, please check.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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 java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
Review Comment:
When you use https://picocli.info/#_inherited_command_attributes feature, you can remove `showDefaultValues` and `scope = INHERIT` in the subcommand.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
Review Comment:
Other commands also have this issue, please check.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1565989921
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java:
##########
@@ -105,192 +103,173 @@ public class PerformanceProducer {
private static IMessageFormatter messageFormatter = null;
- @Command(description = "Test pulsar producer performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceTopicListArguments {
+ @Option(names = { "-threads", "--num-test-threads" }, description = "Number of test threads",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numTestThreads = 1;
- @Option(names = { "-threads", "--num-test-threads" }, description = "Number of test threads",
- converter = PositiveNumberParameterConvert.class
- )
- public int numTestThreads = 1;
+ @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
+ public int msgRate = 100;
- @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
- public int msgRate = 100;
+ @Option(names = { "-s", "--size" }, description = "Message size (bytes)")
+ public int msgSize = 1024;
- @Option(names = { "-s", "--size" }, description = "Message size (bytes)")
- public int msgSize = 1024;
+ @Option(names = { "-n", "--num-producers" }, description = "Number of producers (per topic)",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numProducers = 1;
- @Option(names = { "-n", "--num-producers" }, description = "Number of producers (per topic)",
- converter = PositiveNumberParameterConvert.class
- )
- public int numProducers = 1;
+ @Option(names = {"--separator"}, description = "Separator between the topic and topic number")
+ public String separator = "-";
- @Option(names = {"--separator"}, description = "Separator between the topic and topic number")
- public String separator = "-";
+ @Option(names = {"--send-timeout"}, description = "Set the sendTimeout value default 0 to keep "
+ + "compatibility with previous version of pulsar-perf")
+ public int sendTimeout = 0;
- @Option(names = {"--send-timeout"}, description = "Set the sendTimeout value default 0 to keep "
- + "compatibility with previous version of pulsar-perf")
- public int sendTimeout = 0;
+ @Option(names = { "-pn", "--producer-name" }, description = "Producer Name")
+ public String producerName = null;
- @Option(names = { "-pn", "--producer-name" }, description = "Producer Name")
- public String producerName = null;
+ @Option(names = { "-au", "--admin-url" }, description = "Pulsar Admin URL", descriptionKey = "webServiceUrl")
+ public String adminURL;
- @Option(names = { "-au", "--admin-url" }, description = "Pulsar Admin URL")
- public String adminURL;
+ @Option(names = { "-ch",
+ "--chunking" }, description = "Should split the message and publish in chunks if message size is "
+ + "larger than allowed max size")
+ private boolean chunkingAllowed = false;
- @Option(names = { "-ch",
- "--chunking" }, description = "Should split the message and publish in chunks if message size is "
- + "larger than allowed max size")
- private boolean chunkingAllowed = false;
+ @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding messages")
+ public int maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES;
- @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding messages")
- public int maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES;
+ @Option(names = { "-p", "--max-outstanding-across-partitions" }, description = "Max number of outstanding "
+ + "messages across partitions")
+ public int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
- @Option(names = { "-p", "--max-outstanding-across-partitions" }, description = "Max number of outstanding "
- + "messages across partitions")
- public int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
+ @Option(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;
- @Option(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;
+ @Option(names = { "-m",
+ "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep "
+ + "publishing")
+ public long numMessages = 0;
- @Option(names = { "-m",
- "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep "
- + "publishing")
- public long numMessages = 0;
+ @Option(names = { "-z", "--compression" }, description = "Compress messages payload")
+ public CompressionType compression = CompressionType.NONE;
- @Option(names = { "-z", "--compression" }, description = "Compress messages payload")
- public CompressionType compression = CompressionType.NONE;
+ @Option(names = { "-f", "--payload-file" }, description = "Use payload from an UTF-8 encoded text file and "
+ + "a payload will be randomly selected when publishing messages")
+ public String payloadFilename = null;
- @Option(names = { "-f", "--payload-file" }, description = "Use payload from an UTF-8 encoded text file and "
- + "a payload will be randomly selected when publishing messages")
- public String payloadFilename = null;
+ @Option(names = { "-e", "--payload-delimiter" }, description = "The delimiter used to split lines when "
+ + "using payload from a file")
+ // here escaping \n since default value will be printed with the help text
+ public String payloadDelimiter = "\\n";
- @Option(names = { "-e", "--payload-delimiter" }, description = "The delimiter used to split lines when "
- + "using payload from a file")
- // here escaping \n since default value will be printed with the help text
- public String payloadDelimiter = "\\n";
+ @Option(names = { "-b",
+ "--batch-time-window" }, description = "Batch messages in 'x' ms window (Default: 1ms)")
+ public double batchTimeMillis = 1.0;
- @Option(names = { "-b",
- "--batch-time-window" }, description = "Batch messages in 'x' ms window (Default: 1ms)")
- public double batchTimeMillis = 1.0;
+ @Option(names = { "-db",
+ "--disable-batching" }, description = "Disable batching if true")
+ public boolean disableBatching;
- @Option(names = { "-db",
- "--disable-batching" }, description = "Disable batching if true")
- public boolean disableBatching;
-
- @Option(names = {
+ @Option(names = {
"-bm", "--batch-max-messages"
- }, description = "Maximum number of messages per batch")
- public int batchMaxMessages = DEFAULT_BATCHING_MAX_MESSAGES;
+ }, description = "Maximum number of messages per batch")
+ public int batchMaxMessages = DEFAULT_BATCHING_MAX_MESSAGES;
- @Option(names = {
+ @Option(names = {
"-bb", "--batch-max-bytes"
- }, description = "Maximum number of bytes per batch")
- public int batchMaxBytes = 4 * 1024 * 1024;
+ }, description = "Maximum number of bytes per batch")
+ public int batchMaxBytes = 4 * 1024 * 1024;
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
- @Option(names = "--warmup-time", description = "Warm-up time in seconds (Default: 1 sec)")
- public double warmupTimeSeconds = 1.0;
+ @Option(names = "--warmup-time", description = "Warm-up time in seconds (Default: 1 sec)")
+ public double warmupTimeSeconds = 1.0;
- @Option(names = { "-k", "--encryption-key-name" }, description = "The public key name to encrypt payload")
- public String encKeyName = null;
+ @Option(names = { "-k", "--encryption-key-name" }, description = "The public key name to encrypt payload")
+ public String encKeyName = null;
- @Option(names = { "-v",
- "--encryption-key-value-file" },
- description = "The file which contains the public key to encrypt payload")
- public String encKeyFile = null;
+ @Option(names = { "-v",
+ "--encryption-key-value-file" },
+ description = "The file which contains the public key to encrypt payload")
+ public String encKeyFile = null;
- @Option(names = { "-d",
- "--delay" }, description = "Mark messages with a given delay in seconds")
- public long delay = 0;
+ @Option(names = { "-d",
+ "--delay" }, description = "Mark messages with a given delay in seconds")
+ public long delay = 0;
- @Option(names = { "-dr", "--delay-range"}, description = "Mark messages with a given delay by a random"
- + " number of seconds. this value between the specified origin (inclusive) and the specified bound"
- + " (exclusive). e.g. 1,300", converter = RangeConvert.class)
- public Range<Long> delayRange = null;
+ @Option(names = { "-dr", "--delay-range"}, description = "Mark messages with a given delay by a random"
+ + " number of seconds. this value between the specified origin (inclusive) and the specified bound"
+ + " (exclusive). e.g. 1,300", converter = RangeConvert.class)
+ public Range<Long> delayRange = null;
- @Option(names = { "-set",
- "--set-event-time" }, description = "Set the eventTime on messages")
- public boolean setEventTime = false;
+ @Option(names = { "-set",
+ "--set-event-time" }, description = "Set the eventTime on messages")
+ public boolean setEventTime = false;
- @Option(names = { "-ef",
- "--exit-on-failure" }, description = "Exit from the process on publish failure (default: disable)")
- public boolean exitOnFailure = false;
+ @Option(names = { "-ef",
+ "--exit-on-failure" }, description = "Exit from the process on publish failure (default: disable)")
+ public boolean exitOnFailure = false;
- @Option(names = {"-mk", "--message-key-generation-mode"}, description = "The generation mode of message key"
- + ", valid options are: [autoIncrement, random]")
- public String messageKeyGenerationMode = null;
+ @Option(names = {"-mk", "--message-key-generation-mode"}, description = "The generation mode of message key"
+ + ", valid options are: [autoIncrement, random]", descriptionKey = "messageKeyGenerationMode")
+ public String messageKeyGenerationMode = null;
- @Option(names = { "-am", "--access-mode" }, description = "Producer access mode")
- public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
+ @Option(names = { "-am", "--access-mode" }, description = "Producer access mode")
+ public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
- @Option(names = { "-fp", "--format-payload" },
- description = "Format %%i as a message index in the stream from producer and/or %%t as the timestamp"
- + " nanoseconds.")
- public boolean formatPayload = false;
+ @Option(names = { "-fp", "--format-payload" },
+ description = "Format %%i as a message index in the stream from producer and/or %%t as the timestamp"
+ + " nanoseconds.")
+ public boolean formatPayload = false;
- @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
- public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
+ @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
+ public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
- @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
- + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
- public long transactionTimeout = 10;
+ @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 10;
- @Option(names = {"-nmt", "--numMessage-perTransaction"},
- description = "The number of messages sent by a transaction. "
- + "(After --txn-enable setting to true, -nmt takes effect)")
- public int numMessagesPerTransaction = 50;
+ @Option(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages sent by a transaction. "
+ + "(After --txn-enable setting to true, -nmt takes effect)")
+ public int numMessagesPerTransaction = 50;
- @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
- public boolean isEnableTransaction = false;
+ @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
- @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
- + "setting to true, -abort takes effect)")
- public boolean isAbortTransaction = false;
+ @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
+ + "setting to true, -abort takes effect)")
+ public boolean isAbortTransaction = false;
- @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
- public String histogramFile = null;
+ @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
+ public String histogramFile = null;
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- if (adminURL == null) {
- adminURL = prop.getProperty("webServiceUrl");
- }
- if (adminURL == null) {
- adminURL = prop.getProperty("adminURL", "http://localhost:8080/");
- }
-
- if (isBlank(messageKeyGenerationMode)) {
- messageKeyGenerationMode = prop.getProperty("messageKeyGenerationMode", null);
- }
- }
- }
-
- public static void main(String[] args) throws Exception {
-
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf produce", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java:
##########
@@ -83,325 +82,321 @@ public class PerformanceConsumer {
private static final Recorder recorder = new Recorder(MAX_LATENCY, 5);
private static final Recorder cumulativeRecorder = new Recorder(MAX_LATENCY, 5);
- @Command(description = "Test pulsar consumer performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceTopicListArguments {
+ @Option(names = { "-n", "--num-consumers" }, description = "Number of consumers (per subscription), only "
+ + "one consumer is allowed when subscriptionType is Exclusive",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numConsumers = 1;
- @Option(names = { "-n", "--num-consumers" }, description = "Number of consumers (per subscription), only "
- + "one consumer is allowed when subscriptionType is Exclusive",
- converter = PositiveNumberParameterConvert.class
- )
- public int numConsumers = 1;
+ @Option(names = { "-ns", "--num-subscriptions" }, description = "Number of subscriptions (per topic)",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numSubscriptions = 1;
- @Option(names = { "-ns", "--num-subscriptions" }, description = "Number of subscriptions (per topic)",
- converter = PositiveNumberParameterConvert.class
- )
- public int numSubscriptions = 1;
+ @Option(names = { "-s", "--subscriber-name" }, description = "Subscriber name prefix", hidden = true)
+ public String subscriberName;
- @Option(names = { "-s", "--subscriber-name" }, description = "Subscriber name prefix", hidden = true)
- public String subscriberName;
+ @Option(names = { "-ss", "--subscriptions" },
+ description = "A list of subscriptions to consume (for example, sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
- @Option(names = { "-ss", "--subscriptions" },
- description = "A list of subscriptions to consume (for example, sub1,sub2)")
- public List<String> subscriptions = Collections.singletonList("sub");
+ @Option(names = { "-st", "--subscription-type" }, description = "Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Exclusive;
- @Option(names = { "-st", "--subscription-type" }, description = "Subscription type")
- public SubscriptionType subscriptionType = SubscriptionType.Exclusive;
+ @Option(names = { "-sp", "--subscription-position" }, description = "Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Latest;
- @Option(names = { "-sp", "--subscription-position" }, description = "Subscription position")
- private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Latest;
+ @Option(names = { "-r", "--rate" }, description = "Simulate a slow message consumer (rate in msg/s)")
+ public double rate = 0;
- @Option(names = { "-r", "--rate" }, description = "Simulate a slow message consumer (rate in msg/s)")
- public double rate = 0;
+ @Option(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
- @Option(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
- public int receiverQueueSize = 1000;
+ @Option(names = { "-p", "--receiver-queue-size-across-partitions" },
+ description = "Max total size of the receiver queue across partitions")
+ public int maxTotalReceiverQueueSizeAcrossPartitions = 50000;
- @Option(names = { "-p", "--receiver-queue-size-across-partitions" },
- description = "Max total size of the receiver queue across partitions")
- public int maxTotalReceiverQueueSizeAcrossPartitions = 50000;
+ @Option(names = {"-aq", "--auto-scaled-receiver-queue-size"},
+ description = "Enable autoScaledReceiverQueueSize")
+ public boolean autoScaledReceiverQueueSize = false;
- @Option(names = {"-aq", "--auto-scaled-receiver-queue-size"},
- description = "Enable autoScaledReceiverQueueSize")
- public boolean autoScaledReceiverQueueSize = false;
+ @Option(names = {"-rs", "--replicated" },
+ description = "Whether the subscription status should be replicated")
+ public boolean replicatedSubscription = false;
- @Option(names = {"-rs", "--replicated" },
- description = "Whether the subscription status should be replicated")
- public boolean replicatedSubscription = false;
+ @Option(names = { "--acks-delay-millis" }, description = "Acknowledgements grouping delay in millis")
+ public int acknowledgmentsGroupingDelayMillis = 100;
- @Option(names = { "--acks-delay-millis" }, description = "Acknowledgements grouping delay in millis")
- public int acknowledgmentsGroupingDelayMillis = 100;
+ @Option(names = {"-m",
+ "--num-messages"},
+ description = "Number of messages to consume in total. If <= 0, it will keep consuming")
+ public long numMessages = 0;
- @Option(names = {"-m",
- "--num-messages"},
- description = "Number of messages to consume in total. If <= 0, it will keep consuming")
- public long numMessages = 0;
+ @Option(names = { "-mc", "--max_chunked_msg" }, description = "Max pending chunk messages")
+ private int maxPendingChunkedMessage = 0;
- @Option(names = { "-mc", "--max_chunked_msg" }, description = "Max pending chunk messages")
- private int maxPendingChunkedMessage = 0;
+ @Option(names = { "-ac",
+ "--auto_ack_chunk_q_full" }, description = "Auto ack for oldest message on queue is full")
+ private boolean autoAckOldestChunkedMessageOnQueueFull = false;
- @Option(names = { "-ac",
- "--auto_ack_chunk_q_full" }, description = "Auto ack for oldest message on queue is full")
- private boolean autoAckOldestChunkedMessageOnQueueFull = false;
+ @Option(names = { "-e",
+ "--expire_time_incomplete_chunked_messages" },
+ description = "Expire time in ms for incomplete chunk messages")
+ private long expireTimeOfIncompleteChunkedMessageMs = 0;
- @Option(names = { "-e",
- "--expire_time_incomplete_chunked_messages" },
- description = "Expire time in ms for incomplete chunk messages")
- private long expireTimeOfIncompleteChunkedMessageMs = 0;
+ @Option(names = { "-v",
+ "--encryption-key-value-file" },
+ description = "The file which contains the private key to decrypt payload")
+ public String encKeyFile = null;
- @Option(names = { "-v",
- "--encryption-key-value-file" },
- description = "The file which contains the private key to decrypt payload")
- public String encKeyFile = null;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep consuming")
+ public long testTime = 0;
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep consuming")
- public long testTime = 0;
+ @Option(names = {"--batch-index-ack" }, description = "Enable or disable the batch index acknowledgment")
+ public boolean batchIndexAck = false;
- @Option(names = {"--batch-index-ack" }, description = "Enable or disable the batch index acknowledgment")
- public boolean batchIndexAck = false;
+ @Option(names = { "-pm", "--pool-messages" }, description = "Use the pooled message", arity = "1")
+ private boolean poolMessages = true;
- @Option(names = { "-pm", "--pool-messages" }, description = "Use the pooled message", arity = "1")
- private boolean poolMessages = true;
+ @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 10;
- @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
- + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
- public long transactionTimeout = 10;
+ @Option(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages acknowledged by a transaction. "
+ + "(After --txn-enable setting to true, -numMessage-perTransaction takes effect")
+ public int numMessagesPerTransaction = 50;
- @Option(names = {"-nmt", "--numMessage-perTransaction"},
- description = "The number of messages acknowledged by a transaction. "
- + "(After --txn-enable setting to true, -numMessage-perTransaction takes effect")
- public int numMessagesPerTransaction = 50;
+ @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
- @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
- public boolean isEnableTransaction = false;
+ @Option(names = {"-ntxn"}, description = "The number of opened transactions, 0 means keeping open."
+ + "(After --txn-enable setting to true, -ntxn takes effect.)")
+ public long totalNumTxn = 0;
- @Option(names = {"-ntxn"}, description = "The number of opened transactions, 0 means keeping open."
- + "(After --txn-enable setting to true, -ntxn takes effect.)")
- public long totalNumTxn = 0;
+ @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
+ + "setting to true, -abort takes effect)")
+ public boolean isAbortTransaction = false;
- @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
- + "setting to true, -abort takes effect)")
- public boolean isAbortTransaction = false;
+ @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
+ public String histogramFile = null;
- @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
- public String histogramFile = null;
+ public PerformanceConsumer() {
+ super("consume");
+ }
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- }
- @Override
- public void validate() throws Exception {
- super.validate();
- if (subscriptionType == SubscriptionType.Exclusive && numConsumers > 1) {
- throw new Exception("Only one consumer is allowed when subscriptionType is Exclusive");
- }
+ @Override
+ public void validate() throws Exception {
+ super.validate();
+ if (subscriptionType == SubscriptionType.Exclusive && numConsumers > 1) {
+ throw new Exception("Only one consumer is allowed when subscriptionType is Exclusive");
+ }
- if (subscriptions != null && subscriptions.size() != numSubscriptions) {
- // keep compatibility with the previous version
- if (subscriptions.size() == 1) {
- if (subscriberName == null) {
- subscriberName = subscriptions.get(0);
- }
- List<String> defaultSubscriptions = new ArrayList<>();
- for (int i = 0; i < numSubscriptions; i++) {
- defaultSubscriptions.add(String.format("%s-%d", subscriberName, i));
- }
- subscriptions = defaultSubscriptions;
- } else {
- throw new Exception("The size of subscriptions list should be equal to --num-subscriptions");
+ if (subscriptions != null && subscriptions.size() != numSubscriptions) {
+ // keep compatibility with the previous version
+ if (subscriptions.size() == 1) {
+ if (subscriberName == null) {
+ subscriberName = subscriptions.get(0);
}
+ List<String> defaultSubscriptions = new ArrayList<>();
+ for (int i = 0; i < numSubscriptions; i++) {
+ defaultSubscriptions.add(String.format("%s-%d", subscriberName, i));
+ }
+ subscriptions = defaultSubscriptions;
+ } else {
+ throw new Exception("The size of subscriptions list should be equal to --num-subscriptions");
}
}
}
-
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf consume", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java:
##########
@@ -161,33 +173,40 @@ public void fillArgumentsFromProperties(Properties prop) {
@DataProvider(name = "memoryLimitCliArgumentProvider")
public Object[][] memoryLimitCliArgumentProvider() {
- return new Object[][] {
- { new String[]{"-ml","1"}, 1L},
- { new String[]{"-ml","1K"}, 1024L},
- { new String[]{"--memory-limit", "1G"}, 1024 * 1024 * 1024}
+ return new Object[][]{
+ {new String[]{"-ml", "1"}, 1L},
+ {new String[]{"-ml", "1K"}, 1024L},
+ {new String[]{"--memory-limit", "1G"}, 1024 * 1024 * 1024}
};
}
@Test(dataProvider = "memoryLimitCliArgumentProvider")
- public void testMemoryLimitCliArgument(String[] cliArgs, long expectedMemoryLimit) {
+ public void testMemoryLimitCliArgument(String[] cliArgs, long expectedMemoryLimit) throws Exception {
for (String cmd : List.of(
"pulsar-perf read",
"pulsar-perf produce",
"pulsar-perf consume",
"pulsar-perf transaction"
)) {
// Arrange
- AtomicBoolean called = new AtomicBoolean();
- final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments() {
+ final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments("") {
@Override
- public void fillArgumentsFromProperties(Properties prop) {
- called.set(true);
+ public void run() throws Exception {
+
}
+
};
- baseArgument.confFile = "./src/test/resources/perf_client1.conf";
+ String confFile = "./src/test/resources/perf_client1.conf";
+ Properties prop = new Properties(System.getProperties());
+ try (FileInputStream fis = new FileInputStream(confFile)) {
+ prop.load(fis);
+ }
+ baseArgument.getCommander().setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+ baseArgument.parse(new String[]{});
// Act
- baseArgument.parseCLI(cmd, cliArgs);
+ baseArgument.parseCLI();
Review Comment:
```suggestion
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1565982237
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java:
##########
@@ -90,154 +88,142 @@ public class PerformanceTransaction {
private static final Recorder messageSendRCumulativeRecorder =
new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- @Command(description = "Test pulsar transaction performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceBaseArguments {
+ @Option(names = "--topics-c", description = "All topics that need ack for a transaction", required =
+ true)
+ public List<String> consumerTopic = Collections.singletonList("test-consume");
- @Option(names = "--topics-c", description = "All topics that need ack for a transaction", required =
- true)
- public List<String> consumerTopic = Collections.singletonList("test-consume");
+ @Option(names = "--topics-p", description = "All topics that need produce for a transaction",
+ required = true)
+ public List<String> producerTopic = Collections.singletonList("test-produce");
- @Option(names = "--topics-p", description = "All topics that need produce for a transaction",
- required = true)
- public List<String> producerTopic = Collections.singletonList("test-produce");
+ @Option(names = {"-threads", "--num-test-threads"}, description = "Number of test threads."
+ + "This thread is for a new transaction to ack 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.")
+ public int numTestThreads = 1;
- @Option(names = {"-threads", "--num-test-threads"}, description = "Number of test threads."
- + "This thread is for a new transaction to ack 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.")
- public int numTestThreads = 1;
+ @Option(names = {"-au", "--admin-url"}, description = "Pulsar Admin URL", descriptionKey = "webServiceUrl")
+ public String adminURL;
- @Option(names = {"-au", "--admin-url"}, description = "Pulsar Admin URL")
- public String adminURL;
+ @Option(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with a given number of partitions, 0 means"
+ + "not trying to create a topic")
+ public Integer partitions = null;
- @Option(names = {"-np",
- "--partitions"}, description = "Create partitioned topics with a given number of partitions, 0 means"
- + "not trying to create a topic")
- public Integer partitions = null;
+ @Option(names = {"-time",
+ "--test-duration"}, description = "Test duration (in second). 0 means keeping publishing")
+ public long testTime = 0;
- @Option(names = {"-time",
- "--test-duration"}, description = "Test duration (in second). 0 means keeping publishing")
- public long testTime = 0;
+ @Option(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to consume (for example, sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
- @Option(names = {"-ss",
- "--subscriptions"}, description = "A list of subscriptions to consume (for example, sub1,sub2)")
- public List<String> subscriptions = Collections.singletonList("sub");
+ @Option(names = {"-ns", "--num-subscriptions"}, description = "Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
- @Option(names = {"-ns", "--num-subscriptions"}, description = "Number of subscriptions (per topic)")
- public int numSubscriptions = 1;
+ @Option(names = {"-sp", "--subscription-position"}, description = "Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;
- @Option(names = {"-sp", "--subscription-position"}, description = "Subscription position")
- private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;
+ @Option(names = {"-st", "--subscription-type"}, description = "Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
- @Option(names = {"-st", "--subscription-type"}, description = "Subscription type")
- public SubscriptionType subscriptionType = SubscriptionType.Shared;
+ @Option(names = {"-rs", "--replicated" },
+ description = "Whether the subscription status should be replicated")
+ private boolean replicatedSubscription = false;
- @Option(names = {"-rs", "--replicated" },
- description = "Whether the subscription status should be replicated")
- private boolean replicatedSubscription = false;
+ @Option(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
- @Option(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
- public int receiverQueueSize = 1000;
+ @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 5;
- @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
- + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
- public long transactionTimeout = 5;
+ @Option(names = {"-ntxn",
+ "--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.")
+ public long numTransactions = 0;
- @Option(names = {"-ntxn",
- "--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.")
- public long numTransactions = 0;
+ @Option(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a transaction."
+ + "If transaction disabled, it means the number of messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
- @Option(names = {"-nmp", "--numMessage-perTransaction-produce"},
- description = "Set the number of messages produced in a transaction."
- + "If transaction disabled, it means the number of messages produced in a task.")
- public int numMessagesProducedPerTransaction = 1;
+ @Option(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a transaction."
+ + "If transaction disabled, it means the number of messages consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
- @Option(names = {"-nmc", "--numMessage-perTransaction-consume"},
- description = "Set the number of messages consumed in a transaction."
- + "If transaction disabled, it means the number of messages consumed in a task.")
- public int numMessagesReceivedPerTransaction = 1;
+ @Option(names = {"--txn-disable"}, description = "Disable transaction")
+ public boolean isDisableTransaction = false;
- @Option(names = {"--txn-disable"}, description = "Disable transaction")
- public boolean isDisableTransaction = false;
+ @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-disEnable "
+ + "setting to false, -abort takes effect)")
+ public boolean isAbortTransaction = false;
- @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-disEnable "
- + "setting to false, -abort takes effect)")
- public boolean isAbortTransaction = false;
-
- @Option(names = "-txnRate", description = "Set the rate of opened transaction or task. 0 means no limit")
- public int openTxnRate = 0;
-
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- if (adminURL == null) {
- adminURL = prop.getProperty("webServiceUrl");
- }
- if (adminURL == null) {
- adminURL = prop.getProperty("adminURL", "http://localhost:8080/");
- }
- }
+ @Option(names = "-txnRate", description = "Set the rate of opened transaction or task. 0 means no limit")
+ public int openTxnRate = 0;
+ public PerformanceTransaction() {
+ super("transaction");
}
- public static void main(String[] args)
- throws IOException, PulsarAdminException, ExecutionException, InterruptedException {
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf transaction", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
Unnecessary.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java:
##########
@@ -60,72 +59,65 @@ public class PerformanceReader {
private static Recorder recorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
private static Recorder cumulativeRecorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
- @Command(description = "Test pulsar reader performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceTopicListArguments {
+ @Option(names = {"-r", "--rate"}, description = "Simulate a slow message reader (rate in msg/s)")
+ public double rate = 0;
- @Option(names = { "-r", "--rate" }, description = "Simulate a slow message reader (rate in msg/s)")
- public double rate = 0;
+ @Option(names = {"-m",
+ "--start-message-id"}, description = "Start message id. This can be either 'earliest', "
+ + "'latest' or a specific message id by using 'lid:eid'")
+ public String startMessageId = "earliest";
- @Option(names = { "-m",
- "--start-message-id" }, description = "Start message id. This can be either 'earliest', "
- + "'latest' or a specific message id by using 'lid:eid'")
- public String startMessageId = "earliest";
+ @Option(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
- @Option(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
- public int receiverQueueSize = 1000;
+ @Option(names = {"-n",
+ "--num-messages"}, description = "Number of messages to consume in total. If <= 0, "
+ + "it will keep consuming")
+ public long numMessages = 0;
- @Option(names = {"-n",
- "--num-messages"}, description = "Number of messages to consume in total. If <= 0, "
- + "it will keep consuming")
- public long numMessages = 0;
+ @Option(names = {
+ "--use-tls"}, description = "Use TLS encryption on the connection", descriptionKey = "useTls")
+ public boolean useTls;
- @Option(names = {
- "--use-tls" }, description = "Use TLS encryption on the connection")
- public boolean useTls;
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep consuming")
- public long testTime = 0;
+ @Option(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If <= 0, it will keep consuming")
+ public long testTime = 0;
+ public PerformanceReader() {
+ super("read");
+ }
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- if (!useTls) {
- useTls = Boolean.parseBoolean(prop.getProperty("useTls"));
- }
- }
- @Override
- public void validate() throws Exception {
- super.validate();
- if (startMessageId != "earliest" && startMessageId != "latest"
- && (startMessageId.split(":")).length != 2) {
- String errMsg = String.format("invalid start message ID '%s', must be either either 'earliest', "
- + "'latest' or a specific message id by using 'lid:eid'", startMessageId);
- throw new Exception(errMsg);
- }
+ @Override
+ public void validate() throws Exception {
+ super.validate();
+ if (startMessageId != "earliest" && startMessageId != "latest"
+ && (startMessageId.split(":")).length != 2) {
+ String errMsg = String.format("invalid start message ID '%s', must be either either 'earliest', "
+ + "'latest' or a specific message id by using 'lid:eid'", startMessageId);
+ throw new Exception(errMsg);
}
}
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf read", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566011718
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationClient.java:
##########
@@ -312,54 +306,42 @@ private void handle(final byte command, final DataInputStream inputStream, final
private static final MessageListener<byte[]> ackListener = Consumer::acknowledgeAsync;
/**
- * Create a LoadSimulationClient with the given picocli arguments.
+ * Create a LoadSimulationClient with the given picocli this.
*
- * @param arguments
- * Arguments to configure this from.
*/
- public LoadSimulationClient(final MainArguments arguments) throws Exception {
+ public LoadSimulationClient() throws PulsarClientException {
+ super("simulation-client");
payloadCache = new ConcurrentHashMap<>();
topicsToTradeUnits = new ConcurrentHashMap<>();
-
- admin = PulsarAdmin.builder()
- .serviceHttpUrl(arguments.serviceURL)
- .build();
- client = PulsarClient.builder()
- .memoryLimit(arguments.memoryLimit, SizeUnit.BYTES)
- .serviceUrl(arguments.serviceURL)
- .connectionsPerBroker(4)
- .ioThreads(Runtime.getRuntime().availableProcessors())
- .statsInterval(0, TimeUnit.SECONDS)
- .build();
- port = arguments.port;
- executor = Executors.newCachedThreadPool(new DefaultThreadFactory("test-client"));
}
/**
- * Start a client with command line arguments.
+ * Start a client with command line this.
*
- * @param args
- * Command line arguments to pass in.
*/
- public static void main(String[] args) throws Exception {
- final MainArguments mainArguments = new MainArguments();
- CommandLine commander = new CommandLine(mainArguments);
- commander.setCommandName("pulsar-perf simulation-client");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java:
##########
@@ -541,32 +542,18 @@ private void startBrokerLoadDataStoreMonitor() {
}
}
- /**
- * Run a monitor from command line arguments.
- *
- * @param args Arguments for the monitor.
- */
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- final CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf monitor-brokers");
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1566037656
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java:
##########
@@ -19,53 +19,44 @@
package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isBlank;
-import static org.apache.pulsar.testclient.PerfClientUtils.exit;
-import java.io.File;
-import java.io.FileInputStream;
-import java.util.Properties;
-import lombok.SneakyThrows;
-import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.cli.converters.picocli.ByteUnitToLongConverter;
import org.apache.pulsar.client.api.ProxyProtocol;
-import picocli.CommandLine;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
/**
* PerformanceBaseArguments contains common CLI arguments and parsing logic available to all sub-commands.
* Sub-commands should create Argument subclasses and override the `validate` method as necessary.
*/
-public abstract class PerformanceBaseArguments {
+public abstract class PerformanceBaseArguments extends CmdBase{
Review Comment:
Please move `validate()` to `CmdBase`, add an override in the `PerformanceBaseArguments`, move the `parseCLI` logic to that.
Otherwise, this code will look a bit confusing.
```
public void validate() throws Exception {
}
// Picocli entrypoint.
@Override
public Integer call() throws Exception {
validate();
run();
return 0;
}
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1565986690
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java:
##########
@@ -60,72 +59,65 @@ public class PerformanceReader {
private static Recorder recorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
private static Recorder cumulativeRecorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
- @Command(description = "Test pulsar reader performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceTopicListArguments {
+ @Option(names = {"-r", "--rate"}, description = "Simulate a slow message reader (rate in msg/s)")
+ public double rate = 0;
- @Option(names = { "-r", "--rate" }, description = "Simulate a slow message reader (rate in msg/s)")
- public double rate = 0;
+ @Option(names = {"-m",
+ "--start-message-id"}, description = "Start message id. This can be either 'earliest', "
+ + "'latest' or a specific message id by using 'lid:eid'")
+ public String startMessageId = "earliest";
- @Option(names = { "-m",
- "--start-message-id" }, description = "Start message id. This can be either 'earliest', "
- + "'latest' or a specific message id by using 'lid:eid'")
- public String startMessageId = "earliest";
+ @Option(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
- @Option(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
- public int receiverQueueSize = 1000;
+ @Option(names = {"-n",
+ "--num-messages"}, description = "Number of messages to consume in total. If <= 0, "
+ + "it will keep consuming")
+ public long numMessages = 0;
- @Option(names = {"-n",
- "--num-messages"}, description = "Number of messages to consume in total. If <= 0, "
- + "it will keep consuming")
- public long numMessages = 0;
+ @Option(names = {
+ "--use-tls"}, description = "Use TLS encryption on the connection", descriptionKey = "useTls")
+ public boolean useTls;
- @Option(names = {
- "--use-tls" }, description = "Use TLS encryption on the connection")
- public boolean useTls;
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep consuming")
- public long testTime = 0;
+ @Option(names = {"-time",
+ "--test-duration"}, description = "Test duration in secs. If <= 0, it will keep consuming")
+ public long testTime = 0;
+ public PerformanceReader() {
+ super("read");
+ }
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- if (!useTls) {
- useTls = Boolean.parseBoolean(prop.getProperty("useTls"));
- }
- }
- @Override
- public void validate() throws Exception {
- super.validate();
- if (startMessageId != "earliest" && startMessageId != "latest"
- && (startMessageId.split(":")).length != 2) {
- String errMsg = String.format("invalid start message ID '%s', must be either either 'earliest', "
- + "'latest' or a specific message id by using 'lid:eid'", startMessageId);
- throw new Exception(errMsg);
- }
+ @Override
+ public void validate() throws Exception {
+ super.validate();
+ if (startMessageId != "earliest" && startMessageId != "latest"
+ && (startMessageId.split(":")).length != 2) {
+ String errMsg = String.format("invalid start message ID '%s', must be either either 'earliest', "
+ + "'latest' or a specific message id by using 'lid:eid'", startMessageId);
+ throw new Exception(errMsg);
}
}
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf read", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java:
##########
@@ -161,33 +173,40 @@ public void fillArgumentsFromProperties(Properties prop) {
@DataProvider(name = "memoryLimitCliArgumentProvider")
public Object[][] memoryLimitCliArgumentProvider() {
- return new Object[][] {
- { new String[]{"-ml","1"}, 1L},
- { new String[]{"-ml","1K"}, 1024L},
- { new String[]{"--memory-limit", "1G"}, 1024 * 1024 * 1024}
+ return new Object[][]{
+ {new String[]{"-ml", "1"}, 1L},
+ {new String[]{"-ml", "1K"}, 1024L},
+ {new String[]{"--memory-limit", "1G"}, 1024 * 1024 * 1024}
};
}
@Test(dataProvider = "memoryLimitCliArgumentProvider")
- public void testMemoryLimitCliArgument(String[] cliArgs, long expectedMemoryLimit) {
+ public void testMemoryLimitCliArgument(String[] cliArgs, long expectedMemoryLimit) throws Exception {
for (String cmd : List.of(
"pulsar-perf read",
"pulsar-perf produce",
"pulsar-perf consume",
"pulsar-perf transaction"
)) {
// Arrange
- AtomicBoolean called = new AtomicBoolean();
- final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments() {
+ final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments("") {
@Override
- public void fillArgumentsFromProperties(Properties prop) {
- called.set(true);
+ public void run() throws Exception {
+
}
+
};
- baseArgument.confFile = "./src/test/resources/perf_client1.conf";
+ String confFile = "./src/test/resources/perf_client1.conf";
+ Properties prop = new Properties(System.getProperties());
+ try (FileInputStream fis = new FileInputStream(confFile)) {
+ prop.load(fis);
+ }
+ baseArgument.getCommander().setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+ baseArgument.parse(new String[]{});
// Act
- baseArgument.parseCLI(cmd, cliArgs);
+ baseArgument.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java:
##########
@@ -105,192 +103,173 @@ public class PerformanceProducer {
private static IMessageFormatter messageFormatter = null;
- @Command(description = "Test pulsar producer performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceTopicListArguments {
+ @Option(names = { "-threads", "--num-test-threads" }, description = "Number of test threads",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numTestThreads = 1;
- @Option(names = { "-threads", "--num-test-threads" }, description = "Number of test threads",
- converter = PositiveNumberParameterConvert.class
- )
- public int numTestThreads = 1;
+ @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
+ public int msgRate = 100;
- @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
- public int msgRate = 100;
+ @Option(names = { "-s", "--size" }, description = "Message size (bytes)")
+ public int msgSize = 1024;
- @Option(names = { "-s", "--size" }, description = "Message size (bytes)")
- public int msgSize = 1024;
+ @Option(names = { "-n", "--num-producers" }, description = "Number of producers (per topic)",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numProducers = 1;
- @Option(names = { "-n", "--num-producers" }, description = "Number of producers (per topic)",
- converter = PositiveNumberParameterConvert.class
- )
- public int numProducers = 1;
+ @Option(names = {"--separator"}, description = "Separator between the topic and topic number")
+ public String separator = "-";
- @Option(names = {"--separator"}, description = "Separator between the topic and topic number")
- public String separator = "-";
+ @Option(names = {"--send-timeout"}, description = "Set the sendTimeout value default 0 to keep "
+ + "compatibility with previous version of pulsar-perf")
+ public int sendTimeout = 0;
- @Option(names = {"--send-timeout"}, description = "Set the sendTimeout value default 0 to keep "
- + "compatibility with previous version of pulsar-perf")
- public int sendTimeout = 0;
+ @Option(names = { "-pn", "--producer-name" }, description = "Producer Name")
+ public String producerName = null;
- @Option(names = { "-pn", "--producer-name" }, description = "Producer Name")
- public String producerName = null;
+ @Option(names = { "-au", "--admin-url" }, description = "Pulsar Admin URL", descriptionKey = "webServiceUrl")
+ public String adminURL;
- @Option(names = { "-au", "--admin-url" }, description = "Pulsar Admin URL")
- public String adminURL;
+ @Option(names = { "-ch",
+ "--chunking" }, description = "Should split the message and publish in chunks if message size is "
+ + "larger than allowed max size")
+ private boolean chunkingAllowed = false;
- @Option(names = { "-ch",
- "--chunking" }, description = "Should split the message and publish in chunks if message size is "
- + "larger than allowed max size")
- private boolean chunkingAllowed = false;
+ @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding messages")
+ public int maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES;
- @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding messages")
- public int maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES;
+ @Option(names = { "-p", "--max-outstanding-across-partitions" }, description = "Max number of outstanding "
+ + "messages across partitions")
+ public int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
- @Option(names = { "-p", "--max-outstanding-across-partitions" }, description = "Max number of outstanding "
- + "messages across partitions")
- public int maxPendingMessagesAcrossPartitions = DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
+ @Option(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;
- @Option(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;
+ @Option(names = { "-m",
+ "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep "
+ + "publishing")
+ public long numMessages = 0;
- @Option(names = { "-m",
- "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep "
- + "publishing")
- public long numMessages = 0;
+ @Option(names = { "-z", "--compression" }, description = "Compress messages payload")
+ public CompressionType compression = CompressionType.NONE;
- @Option(names = { "-z", "--compression" }, description = "Compress messages payload")
- public CompressionType compression = CompressionType.NONE;
+ @Option(names = { "-f", "--payload-file" }, description = "Use payload from an UTF-8 encoded text file and "
+ + "a payload will be randomly selected when publishing messages")
+ public String payloadFilename = null;
- @Option(names = { "-f", "--payload-file" }, description = "Use payload from an UTF-8 encoded text file and "
- + "a payload will be randomly selected when publishing messages")
- public String payloadFilename = null;
+ @Option(names = { "-e", "--payload-delimiter" }, description = "The delimiter used to split lines when "
+ + "using payload from a file")
+ // here escaping \n since default value will be printed with the help text
+ public String payloadDelimiter = "\\n";
- @Option(names = { "-e", "--payload-delimiter" }, description = "The delimiter used to split lines when "
- + "using payload from a file")
- // here escaping \n since default value will be printed with the help text
- public String payloadDelimiter = "\\n";
+ @Option(names = { "-b",
+ "--batch-time-window" }, description = "Batch messages in 'x' ms window (Default: 1ms)")
+ public double batchTimeMillis = 1.0;
- @Option(names = { "-b",
- "--batch-time-window" }, description = "Batch messages in 'x' ms window (Default: 1ms)")
- public double batchTimeMillis = 1.0;
+ @Option(names = { "-db",
+ "--disable-batching" }, description = "Disable batching if true")
+ public boolean disableBatching;
- @Option(names = { "-db",
- "--disable-batching" }, description = "Disable batching if true")
- public boolean disableBatching;
-
- @Option(names = {
+ @Option(names = {
"-bm", "--batch-max-messages"
- }, description = "Maximum number of messages per batch")
- public int batchMaxMessages = DEFAULT_BATCHING_MAX_MESSAGES;
+ }, description = "Maximum number of messages per batch")
+ public int batchMaxMessages = DEFAULT_BATCHING_MAX_MESSAGES;
- @Option(names = {
+ @Option(names = {
"-bb", "--batch-max-bytes"
- }, description = "Maximum number of bytes per batch")
- public int batchMaxBytes = 4 * 1024 * 1024;
+ }, description = "Maximum number of bytes per batch")
+ public int batchMaxBytes = 4 * 1024 * 1024;
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
- @Option(names = "--warmup-time", description = "Warm-up time in seconds (Default: 1 sec)")
- public double warmupTimeSeconds = 1.0;
+ @Option(names = "--warmup-time", description = "Warm-up time in seconds (Default: 1 sec)")
+ public double warmupTimeSeconds = 1.0;
- @Option(names = { "-k", "--encryption-key-name" }, description = "The public key name to encrypt payload")
- public String encKeyName = null;
+ @Option(names = { "-k", "--encryption-key-name" }, description = "The public key name to encrypt payload")
+ public String encKeyName = null;
- @Option(names = { "-v",
- "--encryption-key-value-file" },
- description = "The file which contains the public key to encrypt payload")
- public String encKeyFile = null;
+ @Option(names = { "-v",
+ "--encryption-key-value-file" },
+ description = "The file which contains the public key to encrypt payload")
+ public String encKeyFile = null;
- @Option(names = { "-d",
- "--delay" }, description = "Mark messages with a given delay in seconds")
- public long delay = 0;
+ @Option(names = { "-d",
+ "--delay" }, description = "Mark messages with a given delay in seconds")
+ public long delay = 0;
- @Option(names = { "-dr", "--delay-range"}, description = "Mark messages with a given delay by a random"
- + " number of seconds. this value between the specified origin (inclusive) and the specified bound"
- + " (exclusive). e.g. 1,300", converter = RangeConvert.class)
- public Range<Long> delayRange = null;
+ @Option(names = { "-dr", "--delay-range"}, description = "Mark messages with a given delay by a random"
+ + " number of seconds. this value between the specified origin (inclusive) and the specified bound"
+ + " (exclusive). e.g. 1,300", converter = RangeConvert.class)
+ public Range<Long> delayRange = null;
- @Option(names = { "-set",
- "--set-event-time" }, description = "Set the eventTime on messages")
- public boolean setEventTime = false;
+ @Option(names = { "-set",
+ "--set-event-time" }, description = "Set the eventTime on messages")
+ public boolean setEventTime = false;
- @Option(names = { "-ef",
- "--exit-on-failure" }, description = "Exit from the process on publish failure (default: disable)")
- public boolean exitOnFailure = false;
+ @Option(names = { "-ef",
+ "--exit-on-failure" }, description = "Exit from the process on publish failure (default: disable)")
+ public boolean exitOnFailure = false;
- @Option(names = {"-mk", "--message-key-generation-mode"}, description = "The generation mode of message key"
- + ", valid options are: [autoIncrement, random]")
- public String messageKeyGenerationMode = null;
+ @Option(names = {"-mk", "--message-key-generation-mode"}, description = "The generation mode of message key"
+ + ", valid options are: [autoIncrement, random]", descriptionKey = "messageKeyGenerationMode")
+ public String messageKeyGenerationMode = null;
- @Option(names = { "-am", "--access-mode" }, description = "Producer access mode")
- public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
+ @Option(names = { "-am", "--access-mode" }, description = "Producer access mode")
+ public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
- @Option(names = { "-fp", "--format-payload" },
- description = "Format %%i as a message index in the stream from producer and/or %%t as the timestamp"
- + " nanoseconds.")
- public boolean formatPayload = false;
+ @Option(names = { "-fp", "--format-payload" },
+ description = "Format %%i as a message index in the stream from producer and/or %%t as the timestamp"
+ + " nanoseconds.")
+ public boolean formatPayload = false;
- @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
- public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
+ @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
+ public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
- @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
- + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
- public long transactionTimeout = 10;
+ @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 10;
- @Option(names = {"-nmt", "--numMessage-perTransaction"},
- description = "The number of messages sent by a transaction. "
- + "(After --txn-enable setting to true, -nmt takes effect)")
- public int numMessagesPerTransaction = 50;
+ @Option(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages sent by a transaction. "
+ + "(After --txn-enable setting to true, -nmt takes effect)")
+ public int numMessagesPerTransaction = 50;
- @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
- public boolean isEnableTransaction = false;
+ @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
- @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
- + "setting to true, -abort takes effect)")
- public boolean isAbortTransaction = false;
+ @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
+ + "setting to true, -abort takes effect)")
+ public boolean isAbortTransaction = false;
- @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
- public String histogramFile = null;
+ @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
+ public String histogramFile = null;
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- if (adminURL == null) {
- adminURL = prop.getProperty("webServiceUrl");
- }
- if (adminURL == null) {
- adminURL = prop.getProperty("adminURL", "http://localhost:8080/");
- }
-
- if (isBlank(messageKeyGenerationMode)) {
- messageKeyGenerationMode = prop.getProperty("messageKeyGenerationMode", null);
- }
- }
- }
-
- public static void main(String[] args) throws Exception {
-
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf produce", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,153 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(name = "websocket-producer", description = "Test pulsar websocket producer performance.")
+public class PerformanceClient extends CmdBase {
private static final LongAdder messagesSent = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
private static IMessageFormatter messageFormatter = null;
- private CommandLine commander;
- @Command(description = "Test pulsar websocket producer performance.")
- static class Arguments {
+ @Option(names = { "-cf", "--conf-file" }, description = "Configuration file")
+ public String confFile;
- @Option(names = { "-h", "--help" }, description = "Help message", help = true)
- boolean help;
+ @Option(names = { "-u", "--proxy-url" }, description = "Pulsar Proxy URL, e.g., \"ws://localhost:8080/\"")
+ public String proxyURL;
- @Option(names = { "-cf", "--conf-file" }, description = "Configuration file")
- public String confFile;
+ @Parameters(description = "persistent://tenant/ns/my-topic", arity = "1")
+ public List<String> topics;
- @Option(names = { "-u", "--proxy-url" }, description = "Pulsar Proxy URL, e.g., \"ws://localhost:8080/\"")
- public String proxyURL;
+ @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
+ public int msgRate = 100;
- @Parameters(description = "persistent://tenant/ns/my-topic", arity = "1")
- public List<String> topics;
+ @Option(names = { "-s", "--size" }, description = "Message size in byte")
+ public int msgSize = 1024;
- @Option(names = { "-r", "--rate" }, description = "Publish rate msg/s across topics")
- public int msgRate = 100;
+ @Option(names = { "-t", "--num-topic" }, description = "Number of topics",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numTopics = 1;
- @Option(names = { "-s", "--size" }, description = "Message size in byte")
- public int msgSize = 1024;
+ @Option(names = { "--auth_plugin" }, description = "Authentication plugin class name", hidden = true)
+ public String deprecatedAuthPluginClassName;
- @Option(names = { "-t", "--num-topic" }, description = "Number of topics",
- converter = PositiveNumberParameterConvert.class
- )
- public int numTopics = 1;
+ @Option(names = { "--auth-plugin" }, description = "Authentication plugin class name")
+ public String authPluginClassName;
- @Option(names = { "--auth_plugin" }, description = "Authentication plugin class name", hidden = true)
- public String deprecatedAuthPluginClassName;
+ @Option(
+ names = { "--auth-params" },
+ description = "Authentication parameters, whose format is determined by the implementation "
+ + "of method `configure` in authentication plugin class, for example \"key1:val1,key2:val2\" "
+ + "or \"{\"key1\":\"val1\",\"key2\":\"val2\"}\".")
+ public String authParams;
- @Option(names = { "--auth-plugin" }, description = "Authentication plugin class name")
- public String authPluginClassName;
+ @Option(names = { "-m",
+ "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep"
+ + " publishing")
+ public long numMessages = 0;
- @Option(
- names = { "--auth-params" },
- description = "Authentication parameters, whose format is determined by the implementation "
- + "of method `configure` in authentication plugin class, for example \"key1:val1,key2:val2\" "
- + "or \"{\"key1\":\"val1\",\"key2\":\"val2\"}\".")
- public String authParams;
+ @Option(names = { "-f", "--payload-file" }, description = "Use payload from a file instead of empty buffer")
+ public String payloadFilename = null;
- @Option(names = { "-m",
- "--num-messages" }, description = "Number of messages to publish in total. If <= 0, it will keep"
- + " publishing")
- public long numMessages = 0;
+ @Option(names = { "-e", "--payload-delimiter" },
+ description = "The delimiter used to split lines when using payload from a file")
+ // here escaping \n since default value will be printed with the help text
+ public String payloadDelimiter = "\\n";
- @Option(names = { "-f", "--payload-file" }, description = "Use payload from a file instead of empty buffer")
- public String payloadFilename = null;
+ @Option(names = { "-fp", "--format-payload" },
+ description = "Format %i as a message index in the stream from producer and/or %t as the timestamp"
+ + " nanoseconds")
+ public boolean formatPayload = false;
- @Option(names = { "-e", "--payload-delimiter" },
- description = "The delimiter used to split lines when using payload from a file")
- // here escaping \n since default value will be printed with the help text
- public String payloadDelimiter = "\\n";
+ @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
+ public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
- @Option(names = { "-fp", "--format-payload" },
- description = "Format %i as a message index in the stream from producer and/or %t as the timestamp"
- + " nanoseconds")
- public boolean formatPayload = false;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
- @Option(names = {"-fc", "--format-class"}, description = "Custom Formatter class name")
- public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ public PerformanceClient() {
+ super("websocket-producer");
}
- public Arguments loadArguments(String[] args) {
- Arguments arguments = new Arguments();
- commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf websocket-producer");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
- if (arguments.help) {
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ public void loadArguments() {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
@Spec
private CommandSpec commandSpec;
public void loadArguments() {
CommandLine commander = commandSpec.commandLine();
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java:
##########
@@ -90,154 +88,142 @@ public class PerformanceTransaction {
private static final Recorder messageSendRCumulativeRecorder =
new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- @Command(description = "Test pulsar transaction performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceBaseArguments {
+ @Option(names = "--topics-c", description = "All topics that need ack for a transaction", required =
+ true)
+ public List<String> consumerTopic = Collections.singletonList("test-consume");
- @Option(names = "--topics-c", description = "All topics that need ack for a transaction", required =
- true)
- public List<String> consumerTopic = Collections.singletonList("test-consume");
+ @Option(names = "--topics-p", description = "All topics that need produce for a transaction",
+ required = true)
+ public List<String> producerTopic = Collections.singletonList("test-produce");
- @Option(names = "--topics-p", description = "All topics that need produce for a transaction",
- required = true)
- public List<String> producerTopic = Collections.singletonList("test-produce");
+ @Option(names = {"-threads", "--num-test-threads"}, description = "Number of test threads."
+ + "This thread is for a new transaction to ack 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.")
+ public int numTestThreads = 1;
- @Option(names = {"-threads", "--num-test-threads"}, description = "Number of test threads."
- + "This thread is for a new transaction to ack 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.")
- public int numTestThreads = 1;
+ @Option(names = {"-au", "--admin-url"}, description = "Pulsar Admin URL", descriptionKey = "webServiceUrl")
+ public String adminURL;
- @Option(names = {"-au", "--admin-url"}, description = "Pulsar Admin URL")
- public String adminURL;
+ @Option(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with a given number of partitions, 0 means"
+ + "not trying to create a topic")
+ public Integer partitions = null;
- @Option(names = {"-np",
- "--partitions"}, description = "Create partitioned topics with a given number of partitions, 0 means"
- + "not trying to create a topic")
- public Integer partitions = null;
+ @Option(names = {"-time",
+ "--test-duration"}, description = "Test duration (in second). 0 means keeping publishing")
+ public long testTime = 0;
- @Option(names = {"-time",
- "--test-duration"}, description = "Test duration (in second). 0 means keeping publishing")
- public long testTime = 0;
+ @Option(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to consume (for example, sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
- @Option(names = {"-ss",
- "--subscriptions"}, description = "A list of subscriptions to consume (for example, sub1,sub2)")
- public List<String> subscriptions = Collections.singletonList("sub");
+ @Option(names = {"-ns", "--num-subscriptions"}, description = "Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
- @Option(names = {"-ns", "--num-subscriptions"}, description = "Number of subscriptions (per topic)")
- public int numSubscriptions = 1;
+ @Option(names = {"-sp", "--subscription-position"}, description = "Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;
- @Option(names = {"-sp", "--subscription-position"}, description = "Subscription position")
- private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;
+ @Option(names = {"-st", "--subscription-type"}, description = "Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
- @Option(names = {"-st", "--subscription-type"}, description = "Subscription type")
- public SubscriptionType subscriptionType = SubscriptionType.Shared;
+ @Option(names = {"-rs", "--replicated" },
+ description = "Whether the subscription status should be replicated")
+ private boolean replicatedSubscription = false;
- @Option(names = {"-rs", "--replicated" },
- description = "Whether the subscription status should be replicated")
- private boolean replicatedSubscription = false;
+ @Option(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
- @Option(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
- public int receiverQueueSize = 1000;
+ @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 5;
- @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
- + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
- public long transactionTimeout = 5;
+ @Option(names = {"-ntxn",
+ "--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.")
+ public long numTransactions = 0;
- @Option(names = {"-ntxn",
- "--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.")
- public long numTransactions = 0;
+ @Option(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a transaction."
+ + "If transaction disabled, it means the number of messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
- @Option(names = {"-nmp", "--numMessage-perTransaction-produce"},
- description = "Set the number of messages produced in a transaction."
- + "If transaction disabled, it means the number of messages produced in a task.")
- public int numMessagesProducedPerTransaction = 1;
+ @Option(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a transaction."
+ + "If transaction disabled, it means the number of messages consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
- @Option(names = {"-nmc", "--numMessage-perTransaction-consume"},
- description = "Set the number of messages consumed in a transaction."
- + "If transaction disabled, it means the number of messages consumed in a task.")
- public int numMessagesReceivedPerTransaction = 1;
+ @Option(names = {"--txn-disable"}, description = "Disable transaction")
+ public boolean isDisableTransaction = false;
- @Option(names = {"--txn-disable"}, description = "Disable transaction")
- public boolean isDisableTransaction = false;
+ @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-disEnable "
+ + "setting to false, -abort takes effect)")
+ public boolean isAbortTransaction = false;
- @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-disEnable "
- + "setting to false, -abort takes effect)")
- public boolean isAbortTransaction = false;
-
- @Option(names = "-txnRate", description = "Set the rate of opened transaction or task. 0 means no limit")
- public int openTxnRate = 0;
-
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- if (adminURL == null) {
- adminURL = prop.getProperty("webServiceUrl");
- }
- if (adminURL == null) {
- adminURL = prop.getProperty("adminURL", "http://localhost:8080/");
- }
- }
+ @Option(names = "-txnRate", description = "Set the rate of opened transaction or task. 0 means no limit")
+ public int openTxnRate = 0;
+ public PerformanceTransaction() {
+ super("transaction");
}
- public static void main(String[] args)
- throws IOException, PulsarAdminException, ExecutionException, InterruptedException {
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf transaction", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
Unnecessary.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/BrokerMonitor.java:
##########
@@ -541,32 +542,18 @@ private void startBrokerLoadDataStoreMonitor() {
}
}
- /**
- * Run a monitor from command line arguments.
- *
- * @param args Arguments for the monitor.
- */
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- final CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf monitor-brokers");
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java:
##########
@@ -19,53 +19,44 @@
package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isBlank;
-import static org.apache.pulsar.testclient.PerfClientUtils.exit;
-import java.io.File;
-import java.io.FileInputStream;
-import java.util.Properties;
-import lombok.SneakyThrows;
-import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.cli.converters.picocli.ByteUnitToLongConverter;
import org.apache.pulsar.client.api.ProxyProtocol;
-import picocli.CommandLine;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
/**
* PerformanceBaseArguments contains common CLI arguments and parsing logic available to all sub-commands.
* Sub-commands should create Argument subclasses and override the `validate` method as necessary.
*/
-public abstract class PerformanceBaseArguments {
+public abstract class PerformanceBaseArguments extends CmdBase{
Review Comment:
Please move `validate()` to `CmdBase`, add an override in the `PerformanceBaseArguments`, move the `parseCLI` logic to that.
Otherwise, this code will look a bit confusing.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/ManagedLedgerWriter.java:
##########
@@ -79,86 +78,69 @@ public class ManagedLedgerWriter {
private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMillis(120000), 5);
- @Command(description = "Write directly on managed-ledgers", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments {
- @Option(names = { "-h", "--help" }, description = "Help message", help = true)
- boolean help;
+ @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
+ public int msgRate = 100;
- @Option(names = { "-r", "--rate" }, description = "Write rate msg/s across managed ledgers")
- public int msgRate = 100;
+ @Option(names = { "-s", "--size" }, description = "Message size")
+ public int msgSize = 1024;
- @Option(names = { "-s", "--size" }, description = "Message size")
- public int msgSize = 1024;
+ @Option(names = { "-t", "--num-topic" },
+ description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
+ public int numManagedLedgers = 1;
- @Option(names = { "-t", "--num-topic" },
- description = "Number of managed ledgers", converter = PositiveNumberParameterConvert.class)
- public int numManagedLedgers = 1;
+ @Option(names = { "--threads" },
+ description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
+ public int numThreads = 1;
- @Option(names = { "--threads" },
- description = "Number of threads writing", converter = PositiveNumberParameterConvert.class)
- public int numThreads = 1;
+ @Deprecated
+ @Option(names = {"-zk", "--zookeeperServers"},
+ description = "ZooKeeper connection string",
+ hidden = true)
+ public String zookeeperServers;
- @Deprecated
- @Option(names = {"-zk", "--zookeeperServers"},
- description = "ZooKeeper connection string",
- hidden = true)
- public String zookeeperServers;
+ @Option(names = {"-md",
+ "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
+ private String metadataStoreUrl;
- @Option(names = {"-md",
- "--metadata-store"}, description = "Metadata store service URL. For example: zk:my-zk:2181")
- private String metadataStoreUrl;
+ @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
+ public int maxOutstanding = 1000;
- @Option(names = { "-o", "--max-outstanding" }, description = "Max number of outstanding requests")
- public int maxOutstanding = 1000;
+ @Option(names = { "-c",
+ "--max-connections" }, description = "Max number of TCP connections to a single bookie")
+ public int maxConnections = 1;
- @Option(names = { "-c",
- "--max-connections" }, description = "Max number of TCP connections to a single bookie")
- public int maxConnections = 1;
+ @Option(names = { "-m",
+ "--num-messages" },
+ description = "Number of messages to publish in total. If <= 0, it will keep publishing")
+ public long numMessages = 0;
- @Option(names = { "-m",
- "--num-messages" },
- description = "Number of messages to publish in total. If <= 0, it will keep publishing")
- public long numMessages = 0;
+ @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
+ public int ensembleSize = 1;
- @Option(names = { "-e", "--ensemble-size" }, description = "Ledger ensemble size")
- public int ensembleSize = 1;
+ @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
+ public int writeQuorum = 1;
- @Option(names = { "-w", "--write-quorum" }, description = "Ledger write quorum")
- public int writeQuorum = 1;
+ @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
+ public int ackQuorum = 1;
- @Option(names = { "-a", "--ack-quorum" }, description = "Ledger ack quorum")
- public int ackQuorum = 1;
+ @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
+ public DigestType digestType = DigestType.CRC32C;
- @Option(names = { "-dt", "--digest-type" }, description = "BookKeeper digest type")
- public DigestType digestType = DigestType.CRC32C;
-
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
- public long testTime = 0;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep publishing")
+ public long testTime = 0;
+ public ManagedLedgerWriter() {
+ super("managed-ledger");
}
- public static void main(String[] args) throws Exception {
-
- final Arguments arguments = new Arguments();
- CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf managed-ledger");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
-
- if (arguments.help) {
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
Please use the `@Spec`.
This method returns the root commander, not the current commander.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java:
##########
@@ -83,325 +82,321 @@ public class PerformanceConsumer {
private static final Recorder recorder = new Recorder(MAX_LATENCY, 5);
private static final Recorder cumulativeRecorder = new Recorder(MAX_LATENCY, 5);
- @Command(description = "Test pulsar consumer performance.", showDefaultValues = true, scope = ScopeType.INHERIT)
- static class Arguments extends PerformanceTopicListArguments {
+ @Option(names = { "-n", "--num-consumers" }, description = "Number of consumers (per subscription), only "
+ + "one consumer is allowed when subscriptionType is Exclusive",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numConsumers = 1;
- @Option(names = { "-n", "--num-consumers" }, description = "Number of consumers (per subscription), only "
- + "one consumer is allowed when subscriptionType is Exclusive",
- converter = PositiveNumberParameterConvert.class
- )
- public int numConsumers = 1;
+ @Option(names = { "-ns", "--num-subscriptions" }, description = "Number of subscriptions (per topic)",
+ converter = PositiveNumberParameterConvert.class
+ )
+ public int numSubscriptions = 1;
- @Option(names = { "-ns", "--num-subscriptions" }, description = "Number of subscriptions (per topic)",
- converter = PositiveNumberParameterConvert.class
- )
- public int numSubscriptions = 1;
+ @Option(names = { "-s", "--subscriber-name" }, description = "Subscriber name prefix", hidden = true)
+ public String subscriberName;
- @Option(names = { "-s", "--subscriber-name" }, description = "Subscriber name prefix", hidden = true)
- public String subscriberName;
+ @Option(names = { "-ss", "--subscriptions" },
+ description = "A list of subscriptions to consume (for example, sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
- @Option(names = { "-ss", "--subscriptions" },
- description = "A list of subscriptions to consume (for example, sub1,sub2)")
- public List<String> subscriptions = Collections.singletonList("sub");
+ @Option(names = { "-st", "--subscription-type" }, description = "Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Exclusive;
- @Option(names = { "-st", "--subscription-type" }, description = "Subscription type")
- public SubscriptionType subscriptionType = SubscriptionType.Exclusive;
+ @Option(names = { "-sp", "--subscription-position" }, description = "Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Latest;
- @Option(names = { "-sp", "--subscription-position" }, description = "Subscription position")
- private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Latest;
+ @Option(names = { "-r", "--rate" }, description = "Simulate a slow message consumer (rate in msg/s)")
+ public double rate = 0;
- @Option(names = { "-r", "--rate" }, description = "Simulate a slow message consumer (rate in msg/s)")
- public double rate = 0;
+ @Option(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
- @Option(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
- public int receiverQueueSize = 1000;
+ @Option(names = { "-p", "--receiver-queue-size-across-partitions" },
+ description = "Max total size of the receiver queue across partitions")
+ public int maxTotalReceiverQueueSizeAcrossPartitions = 50000;
- @Option(names = { "-p", "--receiver-queue-size-across-partitions" },
- description = "Max total size of the receiver queue across partitions")
- public int maxTotalReceiverQueueSizeAcrossPartitions = 50000;
+ @Option(names = {"-aq", "--auto-scaled-receiver-queue-size"},
+ description = "Enable autoScaledReceiverQueueSize")
+ public boolean autoScaledReceiverQueueSize = false;
- @Option(names = {"-aq", "--auto-scaled-receiver-queue-size"},
- description = "Enable autoScaledReceiverQueueSize")
- public boolean autoScaledReceiverQueueSize = false;
+ @Option(names = {"-rs", "--replicated" },
+ description = "Whether the subscription status should be replicated")
+ public boolean replicatedSubscription = false;
- @Option(names = {"-rs", "--replicated" },
- description = "Whether the subscription status should be replicated")
- public boolean replicatedSubscription = false;
+ @Option(names = { "--acks-delay-millis" }, description = "Acknowledgements grouping delay in millis")
+ public int acknowledgmentsGroupingDelayMillis = 100;
- @Option(names = { "--acks-delay-millis" }, description = "Acknowledgements grouping delay in millis")
- public int acknowledgmentsGroupingDelayMillis = 100;
+ @Option(names = {"-m",
+ "--num-messages"},
+ description = "Number of messages to consume in total. If <= 0, it will keep consuming")
+ public long numMessages = 0;
- @Option(names = {"-m",
- "--num-messages"},
- description = "Number of messages to consume in total. If <= 0, it will keep consuming")
- public long numMessages = 0;
+ @Option(names = { "-mc", "--max_chunked_msg" }, description = "Max pending chunk messages")
+ private int maxPendingChunkedMessage = 0;
- @Option(names = { "-mc", "--max_chunked_msg" }, description = "Max pending chunk messages")
- private int maxPendingChunkedMessage = 0;
+ @Option(names = { "-ac",
+ "--auto_ack_chunk_q_full" }, description = "Auto ack for oldest message on queue is full")
+ private boolean autoAckOldestChunkedMessageOnQueueFull = false;
- @Option(names = { "-ac",
- "--auto_ack_chunk_q_full" }, description = "Auto ack for oldest message on queue is full")
- private boolean autoAckOldestChunkedMessageOnQueueFull = false;
+ @Option(names = { "-e",
+ "--expire_time_incomplete_chunked_messages" },
+ description = "Expire time in ms for incomplete chunk messages")
+ private long expireTimeOfIncompleteChunkedMessageMs = 0;
- @Option(names = { "-e",
- "--expire_time_incomplete_chunked_messages" },
- description = "Expire time in ms for incomplete chunk messages")
- private long expireTimeOfIncompleteChunkedMessageMs = 0;
+ @Option(names = { "-v",
+ "--encryption-key-value-file" },
+ description = "The file which contains the private key to decrypt payload")
+ public String encKeyFile = null;
- @Option(names = { "-v",
- "--encryption-key-value-file" },
- description = "The file which contains the private key to decrypt payload")
- public String encKeyFile = null;
+ @Option(names = { "-time",
+ "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep consuming")
+ public long testTime = 0;
- @Option(names = { "-time",
- "--test-duration" }, description = "Test duration in secs. If <= 0, it will keep consuming")
- public long testTime = 0;
+ @Option(names = {"--batch-index-ack" }, description = "Enable or disable the batch index acknowledgment")
+ public boolean batchIndexAck = false;
- @Option(names = {"--batch-index-ack" }, description = "Enable or disable the batch index acknowledgment")
- public boolean batchIndexAck = false;
+ @Option(names = { "-pm", "--pool-messages" }, description = "Use the pooled message", arity = "1")
+ private boolean poolMessages = true;
- @Option(names = { "-pm", "--pool-messages" }, description = "Use the pooled message", arity = "1")
- private boolean poolMessages = true;
+ @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 10;
- @Option(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
- + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
- public long transactionTimeout = 10;
+ @Option(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages acknowledged by a transaction. "
+ + "(After --txn-enable setting to true, -numMessage-perTransaction takes effect")
+ public int numMessagesPerTransaction = 50;
- @Option(names = {"-nmt", "--numMessage-perTransaction"},
- description = "The number of messages acknowledged by a transaction. "
- + "(After --txn-enable setting to true, -numMessage-perTransaction takes effect")
- public int numMessagesPerTransaction = 50;
+ @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
- @Option(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
- public boolean isEnableTransaction = false;
+ @Option(names = {"-ntxn"}, description = "The number of opened transactions, 0 means keeping open."
+ + "(After --txn-enable setting to true, -ntxn takes effect.)")
+ public long totalNumTxn = 0;
- @Option(names = {"-ntxn"}, description = "The number of opened transactions, 0 means keeping open."
- + "(After --txn-enable setting to true, -ntxn takes effect.)")
- public long totalNumTxn = 0;
+ @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
+ + "setting to true, -abort takes effect)")
+ public boolean isAbortTransaction = false;
- @Option(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
- + "setting to true, -abort takes effect)")
- public boolean isAbortTransaction = false;
+ @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
+ public String histogramFile = null;
- @Option(names = { "--histogram-file" }, description = "HdrHistogram output file")
- public String histogramFile = null;
+ public PerformanceConsumer() {
+ super("consume");
+ }
- @Override
- public void fillArgumentsFromProperties(Properties prop) {
- }
- @Override
- public void validate() throws Exception {
- super.validate();
- if (subscriptionType == SubscriptionType.Exclusive && numConsumers > 1) {
- throw new Exception("Only one consumer is allowed when subscriptionType is Exclusive");
- }
+ @Override
+ public void validate() throws Exception {
+ super.validate();
+ if (subscriptionType == SubscriptionType.Exclusive && numConsumers > 1) {
+ throw new Exception("Only one consumer is allowed when subscriptionType is Exclusive");
+ }
- if (subscriptions != null && subscriptions.size() != numSubscriptions) {
- // keep compatibility with the previous version
- if (subscriptions.size() == 1) {
- if (subscriberName == null) {
- subscriberName = subscriptions.get(0);
- }
- List<String> defaultSubscriptions = new ArrayList<>();
- for (int i = 0; i < numSubscriptions; i++) {
- defaultSubscriptions.add(String.format("%s-%d", subscriberName, i));
- }
- subscriptions = defaultSubscriptions;
- } else {
- throw new Exception("The size of subscriptions list should be equal to --num-subscriptions");
+ if (subscriptions != null && subscriptions.size() != numSubscriptions) {
+ // keep compatibility with the previous version
+ if (subscriptions.size() == 1) {
+ if (subscriberName == null) {
+ subscriberName = subscriptions.get(0);
}
+ List<String> defaultSubscriptions = new ArrayList<>();
+ for (int i = 0; i < numSubscriptions; i++) {
+ defaultSubscriptions.add(String.format("%s-%d", subscriberName, i));
+ }
+ subscriptions = defaultSubscriptions;
+ } else {
+ throw new Exception("The size of subscriptions list should be equal to --num-subscriptions");
}
}
}
-
- public static void main(String[] args) throws Exception {
- final Arguments arguments = new Arguments();
- arguments.parseCLI("pulsar-perf consume", args);
+ @Override
+ public void run() throws Exception {
+ super.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationController.java:
##########
@@ -713,20 +688,22 @@ public void run() throws Exception {
/**
* Start a controller with command line arguments.
*
- * @param args
- * Arguments to pass in.
*/
- public static void main(String[] args) throws Exception {
- final MainArguments arguments = new MainArguments();
- final CommandLine commander = new CommandLine(arguments);
- commander.setCommandName("pulsar-perf simulation-controller");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceBaseArgumentsTest.java:
##########
@@ -196,33 +215,39 @@ public void fillArgumentsFromProperties(Properties prop) {
@DataProvider(name = "invalidMemoryLimitCliArgumentProvider")
public Object[][] invalidMemoryLimitCliArgumentProvider() {
- return new Object[][] {
- { new String[]{"-ml","-1"}},
- { new String[]{"-ml","1C"}},
- { new String[]{"--memory-limit", "1Q"}}
+ return new Object[][]{
+ {new String[]{"-ml", "-1"}},
+ {new String[]{"-ml", "1C"}},
+ {new String[]{"--memory-limit", "1Q"}}
};
}
@Test
- public void testMemoryLimitCliArgumentDefault() {
+ public void testMemoryLimitCliArgumentDefault() throws Exception {
for (String cmd : List.of(
"pulsar-perf read",
"pulsar-perf produce",
"pulsar-perf consume",
"pulsar-perf transaction"
)) {
// Arrange
- AtomicBoolean called = new AtomicBoolean();
- final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments() {
+ final PerformanceBaseArguments baseArgument = new PerformanceBaseArguments("") {
@Override
- public void fillArgumentsFromProperties(Properties prop) {
- called.set(true);
+ public void run() throws Exception {
+
}
+
};
- baseArgument.confFile = "./src/test/resources/perf_client1.conf";
+ String confFile = "./src/test/resources/perf_client1.conf";
+ Properties prop = new Properties(System.getProperties());
+ try (FileInputStream fis = new FileInputStream(confFile)) {
+ prop.load(fis);
+ }
+ baseArgument.getCommander().setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+ baseArgument.parse(new String[]{});
// Act
- baseArgument.parseCLI(cmd, new String[]{});
+ baseArgument.parseCLI();
Review Comment:
```suggestion
```
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationClient.java:
##########
@@ -312,54 +306,42 @@ private void handle(final byte command, final DataInputStream inputStream, final
private static final MessageListener<byte[]> ackListener = Consumer::acknowledgeAsync;
/**
- * Create a LoadSimulationClient with the given picocli arguments.
+ * Create a LoadSimulationClient with the given picocli this.
*
- * @param arguments
- * Arguments to configure this from.
*/
- public LoadSimulationClient(final MainArguments arguments) throws Exception {
+ public LoadSimulationClient() throws PulsarClientException {
+ super("simulation-client");
payloadCache = new ConcurrentHashMap<>();
topicsToTradeUnits = new ConcurrentHashMap<>();
-
- admin = PulsarAdmin.builder()
- .serviceHttpUrl(arguments.serviceURL)
- .build();
- client = PulsarClient.builder()
- .memoryLimit(arguments.memoryLimit, SizeUnit.BYTES)
- .serviceUrl(arguments.serviceURL)
- .connectionsPerBroker(4)
- .ioThreads(Runtime.getRuntime().availableProcessors())
- .statsInterval(0, TimeUnit.SECONDS)
- .build();
- port = arguments.port;
- executor = Executors.newCachedThreadPool(new DefaultThreadFactory("test-client"));
}
/**
- * Start a client with command line arguments.
+ * Start a client with command line this.
*
- * @param args
- * Command line arguments to pass in.
*/
- public static void main(String[] args) throws Exception {
- final MainArguments mainArguments = new MainArguments();
- CommandLine commander = new CommandLine(mainArguments);
- commander.setCommandName("pulsar-perf simulation-client");
- try {
- commander.parseArgs(args);
- } catch (ParameterException e) {
- System.out.println(e.getMessage());
- commander.usage(commander.getOut());
- PerfClientUtils.exit(1);
- }
+ @Override
+ public void run() throws Exception {
+ CommandLine commander = super.getCommander();
Review Comment:
```suggestion
```
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1568319343
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceBaseArguments.java:
##########
@@ -19,53 +19,44 @@
package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isBlank;
-import static org.apache.pulsar.testclient.PerfClientUtils.exit;
-import java.io.File;
-import java.io.FileInputStream;
-import java.util.Properties;
-import lombok.SneakyThrows;
-import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.cli.converters.picocli.ByteUnitToLongConverter;
import org.apache.pulsar.client.api.ProxyProtocol;
-import picocli.CommandLine;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
/**
* PerformanceBaseArguments contains common CLI arguments and parsing logic available to all sub-commands.
* Sub-commands should create Argument subclasses and override the `validate` method as necessary.
*/
-public abstract class PerformanceBaseArguments {
+public abstract class PerformanceBaseArguments extends CmdBase{
Review Comment:
Good suggestion. Done.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#issuecomment-2060517909
> This PR is a bit large and not easy to review. We may need to review multiple times to reach an agreement, thanks.
Yes, most of them are some template type modifications, which require more patience to do code review.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1559614458
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
Review Comment:
Done.
##########
pulsar-testclient/src/main/java/org/apache/pulsar/proxy/socket/client/PerformanceClient.java:
##########
@@ -61,172 +62,161 @@
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
-import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
-public class PerformanceClient {
+@Command(description = "Test pulsar websocket producer performance.")
+public class PerformanceClient extends CmdBase {
private static final LongAdder messagesSent = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
private static IMessageFormatter messageFormatter = null;
- private CommandLine commander;
- @Command(description = "Test pulsar websocket producer performance.")
- static class Arguments {
+ @Option(names = { "-h", "--help" }, description = "Help message", help = true)
+ boolean help;
Review Comment:
Done.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#issuecomment-2057253744
This PR is a bit large and not easy to review. We may need to review multiple times to reach an agreement, thanks.
--
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
Re: [PR] [improve][cli] PIP-343: Refactor pulsar-perf to subcommand [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #22388:
URL: https://github.com/apache/pulsar/pull/22388#discussion_r1569017230
##########
pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PulsarPerfTestTool.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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 org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import java.io.FileInputStream;
+import java.lang.reflect.Constructor;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.pulsar.proxy.socket.client.PerformanceClient;
+import picocli.CommandLine;
+
+@CommandLine.Command(name = "pulsar-perf",
+ scope = CommandLine.ScopeType.INHERIT,
+ mixinStandardHelpOptions = true,
+ showDefaultValues = true
+)
+public class PulsarPerfTestTool {
+
+ protected Map<String, Class<?>> commandMap;
+ protected final CommandLine commander;
+
+ public PulsarPerfTestTool() {
+ this.commander = new CommandLine(this);
+ commandMap = new HashMap<>();
+ }
+
+ private String[] initCommander(String[] args) throws Exception {
+ commandMap.put("produce", PerformanceProducer.class);
+ commandMap.put("consume", PerformanceConsumer.class);
+ commandMap.put("transaction", PerformanceTransaction.class);
+ commandMap.put("read", PerformanceReader.class);
+ commandMap.put("monitor-brokers", BrokerMonitor.class);
+ commandMap.put("simulation-client", LoadSimulationClient.class);
+ commandMap.put("simulation-controller", LoadSimulationController.class);
+ commandMap.put("websocket-producer", PerformanceClient.class);
+ commandMap.put("managed-ledger", ManagedLedgerWriter.class);
+ commandMap.put("gen-doc", CmdGenerateDocumentation.class);
+ if (args.length == 0) {
+ System.out.println("Usage: pulsar-perf CONF_FILE_PATH [options] [command] [command options]");
+ PerfClientUtils.exit(0);
+ }
+ String configFile = args[0];
+ Properties prop = new Properties(System.getProperties());
+ if (configFile != null) {
+ try (FileInputStream fis = new FileInputStream(configFile)) {
+ prop.load(fis);
+ }
+ }
+ commander.setDefaultValueProvider(PulsarPerfTestPropertiesProvider.create(prop));
+
+ for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
+ if (PerformanceBaseArguments.class.isAssignableFrom(c.getValue())){
+ Constructor<?> constructor = c.getValue().getDeclaredConstructor();
+ constructor.setAccessible(true);
+ addCommand(c.getKey(), constructor.newInstance());
+ } else {
+ Constructor<?> constructor = c.getValue().getDeclaredConstructor();
+ constructor.setAccessible(true);
+ addCommand(c.getKey(), constructor.newInstance());
+ }
Review Comment:
```suggestion
Constructor<?> constructor = c.getValue().getDeclaredConstructor();
constructor.setAccessible(true);
addCommand(c.getKey(), constructor.newInstance());
```
--
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