Add fix for backup config owner

.gitignore

@@ -7,7 +7,6 @@ target
 *.pyc
 *.svg
 core/*.json
-*.json
 *.retry
 *.pem
 **/.terraform
@@ -19,3 +18,6 @@ core/*.json
 # IntelliJ artifacts
 .idea/
 *.iml
+
+# VS Code artifacts
+.vscode/

benchmark-framework/pom.xml

@@ -31,7 +31,7 @@
 	<artifactId>benchmark-framework</artifactId>
 
 	<properties>
-		<log4j.version>2.9.1</log4j.version>
+		<log4j.version>2.17.1</log4j.version>
 	</properties>
 
 	<dependencies>
@@ -104,6 +104,12 @@
 			<version>${project.version}</version>
 		</dependency>
 
+		<dependency>
+			<groupId>${project.groupId}</groupId>
+			<artifactId>driver-redpanda</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>${project.groupId}</groupId>
 			<artifactId>driver-rocketmq</artifactId>

benchmark-framework/src/main/java/io/openmessaging/benchmark/Benchmark.java

@@ -63,6 +63,12 @@ static class Arguments {
 
         @Parameter(description = "Workloads", required = true)
         public List<String> workloads;
+
+        @Parameter(names = { "-o", "--output" }, description = "Output", required = false)
+        public String output;
+
+	@Parameter(names = { "-v", "--service-version" }, description = "Optional version of the service being benchmarked, embedded in the final result", required = false)
+	public String serviceVersion;
     }
 
     public static void main(String[] args) throws Exception {
@@ -124,8 +130,17 @@ public static void main(String[] args) throws Exception {
         }
 
         workloads.forEach((workloadName, workload) -> {
+
+            try {
+                workload.validate();
+            } catch (IllegalArgumentException e) {
+                e.printStackTrace();
+                System.exit(-1);
+            }
+
             arguments.drivers.forEach(driverConfig -> {
                 try {
+		    String beginTime = dateFormat.format(new Date());
                     File driverConfigFile = new File(driverConfig);
                     DriverConfiguration driverConfiguration = mapper.readValue(driverConfigFile,
                             DriverConfiguration.class);
@@ -140,9 +155,17 @@ public static void main(String[] args) throws Exception {
                     WorkloadGenerator generator = new WorkloadGenerator(driverConfiguration.name, workload, worker);
 
                     TestResult result = generator.run();
-
-                    String fileName = String.format("%s-%s-%s.json", workloadName, driverConfiguration.name,
-                            dateFormat.format(new Date()));
+		    result.beginTime = beginTime;
+		    result.endTime = dateFormat.format(new Date());
+		    result.version = arguments.serviceVersion;
+
+                    String fileName;
+                    if (arguments.output != null  && arguments.output.length() > 0) {
+                        fileName = arguments.output;
+                    } else {
+                        fileName = String.format("%s-%s-%s.json", workloadName, driverConfiguration.name,
+                                dateFormat.format(new Date()));
+                    }
 
                     log.info("Writing test result into {}", fileName);
                     writer.writeValue(new File(fileName), result);

benchmark-framework/src/main/java/io/openmessaging/benchmark/TestResult.java

@@ -26,6 +26,14 @@
 public class TestResult {
     public String workload;
     public String driver;
+    public String version;
+    public String beginTime;
+    public String endTime;
+    public long messageSize;
+    public int topics;
+    public int partitions;
+    public int producersPerTopic;
+    public int consumersPerTopic;
 
     public List<Double> publishRate = new ArrayList<>();
     public List<Double> consumeRate = new ArrayList<>();
@@ -40,6 +48,15 @@ public class TestResult {
     public List<Double> publishLatency9999pct = new ArrayList<>();
     public List<Double> publishLatencyMax = new ArrayList<>();
 
+    public List<Double> publishDelayLatencyAvg = new ArrayList<>();
+    public List<Long> publishDelayLatency50pct = new ArrayList<>();
+    public List<Long> publishDelayLatency75pct = new ArrayList<>();
+    public List<Long> publishDelayLatency95pct = new ArrayList<>();
+    public List<Long> publishDelayLatency99pct = new ArrayList<>();
+    public List<Long> publishDelayLatency999pct = new ArrayList<>();
+    public List<Long> publishDelayLatency9999pct = new ArrayList<>();
+    public List<Long> publishDelayLatencyMax = new ArrayList<>();
+
     public double aggregatedPublishLatencyAvg;
     public double aggregatedPublishLatency50pct;
     public double aggregatedPublishLatency75pct;
@@ -49,8 +66,19 @@ public class TestResult {
     public double aggregatedPublishLatency9999pct;
     public double aggregatedPublishLatencyMax;
 
+    public double aggregatedPublishDelayLatencyAvg;
+    public long aggregatedPublishDelayLatency50pct;
+    public long aggregatedPublishDelayLatency75pct;
+    public long aggregatedPublishDelayLatency95pct;
+    public long aggregatedPublishDelayLatency99pct;
+    public long aggregatedPublishDelayLatency999pct;
+    public long aggregatedPublishDelayLatency9999pct;
+    public long aggregatedPublishDelayLatencyMax;
+
     public Map<Double, Double> aggregatedPublishLatencyQuantiles = new TreeMap<>();
 
+    public Map<Double, Long> aggregatedPublishDelayLatencyQuantiles = new TreeMap<>();
+
     // End to end latencies (from producer to consumer)
     // Latencies are expressed in milliseconds (without decimals)
 
@@ -73,4 +101,16 @@ public class TestResult {
     public double aggregatedEndToEndLatency999pct;
     public double aggregatedEndToEndLatency9999pct;
     public double aggregatedEndToEndLatencyMax;
+
+    public int getTopics() {
+        return topics;
+    }
+
+    public int getPartitions() {
+        return partitions;
+    }
+
+    public long getMessageSize() {
+        return messageSize;
+    }
 }

benchmark-framework/src/main/java/io/openmessaging/benchmark/Workload.java

@@ -44,13 +44,36 @@ public class Workload {
     public int producerRate;
 
     /**
-     * If the consumer backlog is > 0, the generator will accumulate messages until the requested amount of storage is
-     * retained and then it will start the consumers to drain it.
+     * If the consumer backlog is > 0, the generator will accumulate messages until
+     * the requested amount of storage is retained and then it will start the
+     * consumers to drain it.
      *
-     * The testDurationMinutes will be overruled to allow the test to complete when the consumer has drained all the
-     * backlog and it's on par with the producer
+     * The testDurationMinutes will be overruled to allow the test to complete when
+     * the consumer has drained all the backlog and it's on par with the producer
      */
     public long consumerBacklogSizeGB = 0;
 
+    public int warmupDurationMinutes = 30;
     public int testDurationMinutes;
+
+    /**
+     * Consumer only tests. The warm-up phase will be run w/ the specified producer
+     * configuration until the specified backlog is built. Following that, the
+     * producers are shut off and the consumers are run until the backlog is drained
+     */
+    public boolean consumerOnly;
+
+    public void validate() throws IllegalArgumentException {
+        if (consumerOnly && (subscriptionsPerTopic == 0 || consumerPerSubscription == 0)) {
+            throw new IllegalArgumentException("Consumer only tests need subscriptions/consumers");
+        }
+
+        if (consumerOnly && consumerBacklogSizeGB <= 0) {
+            throw new IllegalArgumentException("Consumer only tests need a backlog specification");
+        }
+
+        if (producerRate < 0) {
+            throw new IllegalArgumentException("Producer rate should be > 0");
+        }
+    }
 }