Create CSV file on customized path

README.md

@@ -248,14 +248,15 @@ Run the benchmark from source
   The time units can be "days", "day", "h", "m", "s", "ms", "us", "ns".
   e.g. "--run.until 1m" or "--run.until 89242records" Default: 1000records
 8. --record.size: the (bound of) record size in byte. Default: 1 KiB
-9. --prop.file: the path to property file.
-10. --partitioner: the partitioner to use in producers
-11. --createCSV: put the metrics into a csv file if this flag is set. Default: false
-12. --compression: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd. Default: (NONE)
-13. --jmx.servers: the jmx server addresses of the brokers 
-14. --key.distribution: name of the distribution on key. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: (No key)
-15. --size.distribution: name of the distribution on value size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: "uniform"
-16. --specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
+9. --fixed.size: the flag to let all records have the same size
+10. --prop.file: the path to property file.
+11. --partitioner: the partitioner to use in producers
+12. --jmx.servers: the jmx server addresses of the brokers 
+13. --key.distribution: name of the distribution on key. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: (No key)
+14. --size.distribution: name of the distribution on value size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: "uniform"
+15. --specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
+16. --report.path: A path to place the report file. Default: (no report)
+17. --report.format: Select output file format. Available format: "csv", "json".
 
 ---
 

app/src/main/java/org/astraea/argument/PathField.java

@@ -0,0 +1,11 @@
+package org.astraea.argument;
+
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+
+public class PathField extends Field<Path> {
+  @Override
+  public Path convert(String value) {
+    return FileSystems.getDefault().getPath(value);
+  }
+}

app/src/main/java/org/astraea/performance/Performance.java

@@ -2,7 +2,9 @@
 
 import com.beust.jcommander.Parameter;
 import java.io.IOException;
+import java.nio.file.Path;
 import java.time.Duration;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -23,6 +25,7 @@
 import org.astraea.argument.CompressionField;
 import org.astraea.argument.NonEmptyStringField;
 import org.astraea.argument.NonNegativeShortField;
+import org.astraea.argument.PathField;
 import org.astraea.argument.PositiveLongField;
 import org.astraea.argument.PositiveShortField;
 import org.astraea.argument.StringMapField;
@@ -95,6 +98,11 @@ public static Result execute(final Argument param)
 
     var manager = new Manager(param, producerMetrics, consumerMetrics);
     var tracker = new Tracker(producerMetrics, consumerMetrics, manager);
+    Collection<Executor> fileWriter =
+        (param.CSVPath != null)
+            ? List.of(
+                ReportFormat.createFileWriter(param.reportFormat, param.CSVPath, manager, tracker))
+            : List.of();
     var groupId = "groupId-" + System.currentTimeMillis();
     try (var threadPool =
         ThreadPool.builder()
@@ -133,7 +141,7 @@ public static Result execute(final Argument param)
                                 manager))
                     .collect(Collectors.toUnmodifiableList()))
             .executor(tracker)
-            .executor((param.createCSV) ? new FileWriter(manager, tracker) : () -> State.DONE)
+            .executors(fileWriter)
             .build()) {
       threadPool.waitAll();
       return new Result(param.topic);
@@ -341,11 +349,6 @@ public Map<String, Object> consumerProps() {
       return props;
     }
 
-    @Parameter(
-        names = {"--createCSV"},
-        description = "put the metrics into a csv file if this flag is set")
-    boolean createCSV = false;
-
     @Parameter(
         names = {"--compression"},
         description =
@@ -383,6 +386,18 @@ public boolean transaction() {
             "String: Used with SpecifyBrokerPartitioner to specify the brokers that partitioner can send.",
         validateWith = NonEmptyStringField.class)
     List<Integer> specifyBroker = List.of(-1);
+
+    @Parameter(
+        names = {"--report.path"},
+        description = "String: A path to place the report. Default: (no report)",
+        converter = PathField.class)
+    Path CSVPath = null;
+
+    @Parameter(
+        names = {"--report.format"},
+        description = "Output format for the report",
+        converter = ReportFormat.ReportFormatConverter.class)
+    ReportFormat reportFormat = ReportFormat.CSV;
   }
 
   public static class Result {