Repository: skiptests/astraea
Pull request #225 (2022-02-10 10:17:37)

Argument check strictly
#221

補齊文檔，並做更嚴格的參數檢查。

修改輸出文字的不一致(大小寫、空格)。


Patch:
diff --git a/README.md b/README.md
index ecdcf90f..68505e1a 100644
--- a/README.md
+++ b/README.md
@@ -215,7 +215,10 @@ Run the benchmark from source
 4. --replicas: topic config when creating new topic. Default: 1
 5. --consumers: the number of consumers (threads). Default: 1
 6. --producers: the number of producers (threads). Default: 1
-7. --run.until: the total number of records sent by the producers. Default: 1000records
+7. --run.until: the total number of records sent by the producers or the time for producer to send records.
+  The duration formats accepted are (a number) + (a time unit). 
+  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. --fixed.size: the flag to let all records have the same size
 10. --prop.file: the path to property file.
diff --git a/app/src/main/java/org/astraea/performance/ExeTime.java b/app/src/main/java/org/astraea/performance/ExeTime.java
index 225e255c..6ffbd8fc 100644
--- a/app/src/main/java/org/astraea/performance/ExeTime.java
+++ b/app/src/main/java/org/astraea/performance/ExeTime.java
@@ -1,8 +1,11 @@
 package org.astraea.performance;
 
+import com.beust.jcommander.IParameterValidator;
 import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.ParameterException;
 import java.time.Duration;
 import java.util.function.BiFunction;
+import java.util.regex.Pattern;
 import org.astraea.argument.ArgumentUtil;
 
 /**
@@ -42,4 +45,16 @@ interface ExeTime {
       return ExeTime.of(value);
     }
   }
+
+  class Validator implements IParameterValidator {
+    static final Pattern PATTERN = Pattern.compile("^([0-9]+)(days|day|h|m|s|ms|us|ns|records)$");
+
+    @Override
+    public void validate(String name, String value) throws ParameterException {
+      if (!PATTERN.matcher(value).matches()) {
+        throw new ParameterException(
+            "Invalid ExeTime format. valid format example: \"1m\" or \"89242records\"");
+      }
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 8846a60b..cca22e18 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -5,7 +5,12 @@ import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import java.io.IOException;
 import java.time.Duration;
-import java.util.*;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -15,8 +20,10 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.record.CompressionType;
+import org.astraea.Utils;
 import org.astraea.argument.ArgumentUtil;
 import org.astraea.argument.BasicArgumentWithPropFile;
 import org.astraea.concurrent.ThreadPool;
@@ -71,6 +78,7 @@ public class Performance {
           .topic(param.topic)
           .create();
 
+      Utils.waitFor(() -> topicAdmin.topicNames().contains(param.topic));
       partitions = partition(param, topicAdmin);
     }
 
@@ -212,7 +220,7 @@ public class Performance {
       return topicAdmin
           .partitionsOfBrokers(Set.of(param.topic), new HashSet<>(param.specifyBroker))
           .stream()
-          .map(topicPartition -> topicPartition.partition())
+          .map(TopicPartition::partition)
           .collect(Collectors.toList());
     } else return List.of(-1);
   }
@@ -259,8 +267,9 @@ public class Performance {
         description =
             "Run until number of records are produced and consumed or until duration meets."
                 + " The duration formats accepted are (a number) + (a time unit)."
-                + " The time units can be \"days\", \"day\", \"h\", \"m\", \"s, \"ms\","
+                + " The time units can be \"days\", \"day\", \"h\", \"m\", \"s\", \"ms\","
                 + " \"us\", \"ns\"",
+        validateWith = ExeTime.Validator.class,
         converter = ExeTime.Converter.class)
     ExeTime exeTime = ExeTime.of("1000records");
 
@@ -324,10 +333,15 @@ public class Performance {
 
   static class CompressionArgument implements IStringConverter<CompressionType> {
 
+    /**
+     * @param value Name of compression type. Accept lower-case name only ("none", "gzip", "snappy",
+     *     "lz4", "zstd").
+     */
     @Override
     public CompressionType convert(String value) {
       try {
-        return CompressionType.forName(value.toLowerCase());
+        // `CompressionType#forName` accept lower-case name only.
+        return CompressionType.forName(value);
       } catch (IllegalArgumentException e) {
         throw new ParameterException(
             "the "
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index ab12bc1a..f9b76ddb 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -61,16 +61,16 @@ public class Tracker implements ThreadPool.Executor {
             + duration.toSecondsPart()
             + "sec");
     System.out.printf("producers completion rate: %.2f%%%n", percentage);
-    System.out.printf("  average Throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf("  average throughput: %.3f MB/second%n", result.averageBytes(duration));
     System.out.printf(
         "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
-    System.out.println("  publish max latency: " + result.maxLatency + "ms");
-    System.out.println("  publish mim latency: " + result.minLatency + "ms");
+    System.out.println("  publish max latency: " + result.maxLatency + " ms");
+    System.out.println("  publish mim latency: " + result.minLatency + " ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
           "  producer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
-          "  producer[%d] average publish latency: %.3fms%n", i, result.averageLatencies.get(i));
+          "  producer[%d] average publish latency: %.3f ms%n", i, result.averageLatencies.get(i));
     }
     System.out.println("\n");
     return percentage >= 100D;
@@ -88,13 +88,14 @@ public class Tracker implements ThreadPool.Executor {
     System.out.printf("  average throughput: %.3f MB/second%n", result.averageBytes(duration));
     System.out.printf(
         "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
-    System.out.println("  end-to-end max latency: " + result.maxLatency + "ms");
-    System.out.println("  end-to-end mim latency: " + result.minLatency + "ms");
+    System.out.println("  end-to-end max latency: " + result.maxLatency + " ms");
+    System.out.println("  end-to-end mim latency: " + result.minLatency + " ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
           "  consumer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
-          "  consumer[%d] average ene-to-end latency: %.3fms%n", i, result.averageLatencies.get(i));
+          "  consumer[%d] average ene-to-end latency: %.3f ms%n",
+          i, result.averageLatencies.get(i));
     }
     System.out.println("\n");
     // Target number of records consumed OR consumed all that produced
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index f04c609c..f12ff49a 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -66,7 +66,7 @@ function checkNetwork() {
 }
 
 function rejectProperty() {
-  local key=$1c
+  local key=$1
   if [[ -f "$BROKER_PROPERTIES" ]] && [[ "$(cat $BROKER_PROPERTIES | grep $key)" != "" ]]; then
     echo "$key is NOT configurable"
     exit 2


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/ExeTimeTest.java b/app/src/test/java/org/astraea/performance/ExeTimeTest.java
index 6446d0c9..04cd1e0e 100644
--- a/app/src/test/java/org/astraea/performance/ExeTimeTest.java
+++ b/app/src/test/java/org/astraea/performance/ExeTimeTest.java
@@ -1,13 +1,16 @@
 package org.astraea.performance;
 
+import com.beust.jcommander.ParameterException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class ExeTimeTest {
   private final ExeTime.Converter converter = new ExeTime.Converter();
+  private final ExeTime.Validator validator = new ExeTime.Validator();
 
   @Test
   void testRecords() {
+    Assertions.assertDoesNotThrow(() -> validator.validate("--run.until", "1000records"));
     var exeTime = converter.convert("1000records");
     Assertions.assertEquals(0, exeTime.percentage(0, 10));
     Assertions.assertEquals(100D, exeTime.percentage(1000, 10));
@@ -15,6 +18,7 @@ public class ExeTimeTest {
 
   @Test
   void testDuration() {
+    Assertions.assertDoesNotThrow(() -> validator.validate("--run.until", "100ms"));
     var exeTime = converter.convert("100ms");
     Assertions.assertEquals(0, exeTime.percentage(1000, 0));
     Assertions.assertEquals(100D, exeTime.percentage(1000, 100));
@@ -22,7 +26,13 @@ public class ExeTimeTest {
 
   @Test
   void testUnknownArgument() {
-    Assertions.assertThrows(IllegalArgumentException.class, () -> converter.convert("aaa"));
-    Assertions.assertThrows(IllegalArgumentException.class, () -> converter.convert("10record"));
+    Assertions.assertThrows(
+        ParameterException.class, () -> validator.validate("--run.until", "aaa"));
+    Assertions.assertThrows(
+        ParameterException.class, () -> validator.validate("--run.until", "10record"));
+    Assertions.assertThrows(
+        ParameterException.class, () -> validator.validate("--run.until", "1000"));
+    Assertions.assertThrows(
+        ParameterException.class, () -> validator.validate("--run.until", "10Ms"));
   }
 }