Create CSV file on customized path

README.md

@@ -225,6 +225,8 @@ Run the benchmark from source
 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. Available distribution names: "uniform", "zipfian", "latest". Default: (No key)
+14. --report.path: A path to place the CSV file. Default: (no report)
+15. --report.format: Select output file format. Available format: "csv", "json".
 
 ---
 

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

@@ -3,7 +3,9 @@
 import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.converters.PathConverter;
 import java.io.IOException;
+import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.HashSet;
@@ -98,7 +100,10 @@ public static Future<String> execute(final Argument param)
     var manager = new Manager(param, producerMetrics, consumerMetrics);
     var tracker = new Tracker(producerMetrics, consumerMetrics, manager);
     Collection<ThreadPool.Executor> fileWriter =
-        (param.createCSV) ? List.of(new FileWriter(manager, tracker)) : List.of();
+        (param.CSVPath != null)
+            ? List.of(
+                ReportWriter.createFileWriter(param.reportFormat, param.CSVPath, manager, tracker))
+            : List.of();
     var groupId = "groupId-" + System.currentTimeMillis();
     try (ThreadPool threadPool =
         ThreadPool.builder()
@@ -308,11 +313,6 @@ public Map<String, Object> producerProps() {
       return props;
     }
 
-    @Parameter(
-        names = {"--createCSV"},
-        description = "create the metrics into a csv file if this flag is set")
-    boolean createCSV = false;
-
     @Parameter(
         names = {"--compression"},
         description =
@@ -333,6 +333,18 @@ public Map<String, Object> producerProps() {
             "String: Used with SpecifyBrokerPartitioner to specify the brokers that partitioner can send.",
         validateWith = NotEmptyString.class)
     List<Integer> specifyBroker = List.of(-1);
+
+    @Parameter(
+        names = {"--report.path"},
+        description = "String: A path to place the report. Default: (no report)",
+        converter = PathConverter.class)
+    Path CSVPath = null;
+
+    @Parameter(
+        names = {"--report.format"},
+        description = "Output format for the report",
+        converter = ReportWriter.FileFormat.FileFormatConverter.class)
+    ReportWriter.FileFormat reportFormat = ReportWriter.FileFormat.CSV;
   }
 
   static class CompressionArgument implements IStringConverter<CompressionType> {