Improve perf producer to publish non-empty buffer and multi-lines pay…

…load. (apache#4479) ### Motivation Fixes apache#4478 ### Modifications Generate alphabetic bytes buffer and add a parameter `--payload-delimiter` to select different lines randomly from a file.
vzhikserg · Jun 6, 2019 · c540e28 · c540e28
1 parent b7a5677
commit c540e28
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 5 deletions.
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -35,13 +35,16 @@
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.text.DecimalFormat;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Random;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -136,8 +139,13 @@ static class Arguments {
         @Parameter(names = { "-z", "--compression" }, description = "Compress messages payload")
         public CompressionType compression = CompressionType.NONE;
 
-        @Parameter(names = { "-f", "--payload-file" }, description = "Use payload from a file instead of empty buffer")
+        @Parameter(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;
+
+        @Parameter(names = { "-e", "--payload-delimiter" }, description = "The delimiter used to split lines when using payload from a file")
+        public String payloadDelimiter = "\\n"; // here escaping \n since default value will be printed with the help text
+
         @Parameter(names = { "-b",
                 "--batch-time-window" }, description = "Batch messages in 'x' ms window (Default: 1ms)")
         public double batchTimeMillis = 1.0;
@@ -226,11 +234,25 @@ public static void main(String[] args) throws Exception {
         log.info("Starting Pulsar perf producer with config: {}", w.writeValueAsString(arguments));
 
         // Read payload data from file if needed
-        byte payloadData[];
+        final byte[] payloadBytes = new byte[arguments.msgSize];
+        Random random = new Random(0);
+        List<byte[]> payloadByteList = Lists.newArrayList();
         if (arguments.payloadFilename != null) {
-            payloadData = Files.readAllBytes(Paths.get(arguments.payloadFilename));
+            Path payloadFilePath = Paths.get(arguments.payloadFilename);
+            if (Files.notExists(payloadFilePath) || Files.size(payloadFilePath) == 0)  {
+                throw new IllegalArgumentException("Payload file doesn't exist or it is empty.");
+            }
+            // here escaping the default payload delimiter to correct value
+            String delimiter = arguments.payloadDelimiter.equals("\\n") ? "\n" : arguments.payloadDelimiter;
+            String[] payloadList = new String(Files.readAllBytes(payloadFilePath), StandardCharsets.UTF_8).split(delimiter);
+            log.info("Reading payloads from {} and {} records read", payloadFilePath.toAbsolutePath(), payloadList.length);
+            for (String payload : payloadList) {
+                payloadByteList.add(payload.getBytes(StandardCharsets.UTF_8));
+            }
         } else {
-            payloadData = new byte[arguments.msgSize];
+            for (int i = 0; i < payloadBytes.length; ++i) {
+                payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+            }
         }
 
         // Now processing command line arguments
@@ -359,6 +381,14 @@ public void run() {
 
                         final long sendTime = System.nanoTime();
 
+                        byte[] payloadData;
+
+                        if (arguments.payloadFilename != null) {
+                            payloadData = payloadByteList.get(random.nextInt(payloadByteList.size()));
+                        } else {
+                            payloadData = payloadBytes;
+                        }
+
                         TypedMessageBuilder<byte[]> messageBuilder = producer.newMessage()
                                 .value(payloadData);
                         if (arguments.delay >0) {

diff --git a/site2/docs/reference-cli-tools.md b/site2/docs/reference-cli-tools.md
@@ -455,7 +455,8 @@ Options
 |`-m`, `--num-messages`|Number of messages to publish in total. If set to 0, it will keep publishing.|0|
 |`-n`, `--num-producers`|The number of producers (per topic)|1|
 |`-t`, `--num-topic`|The number of topics|1|
-|`-f`, `--payload-file`|Use payload from a file instead of an empty buffer||
+|`-f`, `--payload-file`|Use payload from an UTF-8 encoded text file and a payload will be randomly selected when publishing messages||
+|`-e`, `--payload-delimiter`|The delimiter used to split lines when using payload from a file|\n|
 |`-r`, `--rate`|Publish rate msg/s across topics|100|
 |`-u`, `--service-url`|Pulsar service URL||
 |`-s`, `--size`|Message size (in bytes)|1024|