Repository: skiptests/astraea
Pull request #161 (2021-12-23 12:13:40)

Use `DataSize` to rewrite `--record.size` for performance tool
為`DataSize`新增字串轉`DataSize`的方法，
並把Performance tool中的參數`--record.size`，改使用`DataSize`表示。(#155)


Patch:
diff --git a/.gitignore b/.gitignore
index c947be9a..3d5abcc6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,4 +15,3 @@ zookeeper.dockerfile
 broker.dockerfile
 kafka_tool.dockerfile
 spark.dockerfile
-spark-job.dockerfile
\ No newline at end of file
diff --git a/README.md b/README.md
index cea4b877..5f7ac3da 100644
--- a/README.md
+++ b/README.md
@@ -116,7 +116,7 @@ Run the benchmark from source
 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
-8. --record.size: the (bound of) record size in byte. Default: 1024 byte
+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.
 11. --partitioner: the partitioner to use in producers
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index 0d737a60..e84b809f 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -6,6 +6,8 @@ import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import org.astraea.utils.DataSize;
+import org.astraea.utils.DataUnit;
 
 /**
  * Thread safe This class is used for managing the start/end of the producer/consumer threads.
@@ -15,7 +17,7 @@ import java.util.concurrent.atomic.AtomicLong;
 public class Manager {
   private final ExeTime exeTime;
   private final boolean fixedSize;
-  private final int size;
+  private final DataSize dataSize;
   private final CountDownLatch getAssignment;
   private final AtomicInteger producerClosed;
   private final Random rand = new Random();
@@ -42,7 +44,10 @@ public class Manager {
   public Manager(
       Performance.Argument argument, List<Metrics> producerMetrics, List<Metrics> consumerMetrics) {
     this.fixedSize = argument.fixedSize;
-    this.size = argument.recordSize;
+    this.dataSize = argument.recordSize;
+    if (dataSize.greaterThan(DataUnit.Byte.of(Integer.MAX_VALUE)))
+      throw new IllegalArgumentException(
+          "Record size should be smaller than or equal to 2147483648 (Integer.MAX_VALUE) bytes");
     this.getAssignment = new CountDownLatch(argument.consumers);
     this.producerClosed = new AtomicInteger(argument.producers);
     this.producerMetrics = producerMetrics;
@@ -61,7 +66,10 @@ public class Manager {
     if (exeTime.percentage(payloadNum.getAndIncrement(), System.currentTimeMillis() - start)
         >= 100D) return Optional.empty();
 
-    byte[] payload = (this.fixedSize) ? new byte[size] : new byte[rand.nextInt(size) + 1];
+    byte[] payload =
+        (this.fixedSize)
+            ? new byte[dataSize.measurement(DataUnit.Byte).intValue()]
+            : new byte[rand.nextInt(dataSize.measurement(DataUnit.Byte).intValue()) + 1];
     rand.nextBytes(payload);
     return Optional.of(payload);
   }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 24c34d82..7f5e1109 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -22,6 +22,8 @@ import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.producer.Producer;
 import org.astraea.topic.TopicAdmin;
+import org.astraea.utils.DataSize;
+import org.astraea.utils.DataUnit;
 
 /**
  * Performance benchmark which includes
@@ -239,9 +241,9 @@ public class Performance {
 
     @Parameter(
         names = {"--record.size"},
-        description = "Integer: size of each record",
-        validateWith = ArgumentUtil.PositiveLong.class)
-    int recordSize = 1024;
+        description = "DataSize: size of each record. e.g. \"500KiB\"",
+        converter = DataSize.Converter.class)
+    DataSize recordSize = DataUnit.KiB.of(1);
 
     @Parameter(
         names = {"--jmx.servers"},
diff --git a/app/src/main/java/org/astraea/utils/DataSize.java b/app/src/main/java/org/astraea/utils/DataSize.java
index a30a9ab7..0fdfb90c 100644
--- a/app/src/main/java/org/astraea/utils/DataSize.java
+++ b/app/src/main/java/org/astraea/utils/DataSize.java
@@ -1,5 +1,6 @@
 package org.astraea.utils;
 
+import com.beust.jcommander.IStringConverter;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.math.MathContext;
@@ -7,6 +8,8 @@ import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Comparator;
 import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /** Data size class */
 public class DataSize implements Comparable<DataSize> {
@@ -208,4 +211,34 @@ public class DataSize implements Comparable<DataSize> {
   public int hashCode() {
     return Objects.hash(bits);
   }
+
+  public static class Converter implements IStringConverter<DataSize> {
+    // Parse number and DataUnit
+    private static final Pattern DATA_SIZE_PATTERN =
+        Pattern.compile("(?<measurement>[0-9]+)\\s?(?<dataUnit>[a-zA-Z]+)");
+
+    /**
+     * Convert string to DataSize.
+     *
+     * <pre>{@code
+     * new DataSize.Converter().convert("500KB");  // 500 KB  (500 * 1000 bytes)
+     * new DataSize.Converter().convert("500KiB"); // 500 KiB (500 * 1024 bytes)
+     * new DataSize.Converter().convert("500Kb");  // 500 Kb  (500 * 1000 bits)
+     * new DataSize.Converter().convert("500Kib"); // 500 Kib (500 * 1024 bits)
+     * }</pre>
+     *
+     * @param argument number and the unit. e.g. "500MiB", "9876 KB"
+     * @return a data size object of given measurement under specific data unit.
+     */
+    @Override
+    public DataSize convert(String argument) {
+      Matcher matcher = DATA_SIZE_PATTERN.matcher(argument);
+      if (matcher.matches()) {
+        return DataUnit.valueOf(matcher.group("dataUnit"))
+            .of(Long.parseLong(matcher.group("measurement")));
+      } else {
+        throw new IllegalArgumentException("Unknown DataSize \"" + argument + "\"");
+      }
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/utils/DataUnit.java b/app/src/main/java/org/astraea/utils/DataUnit.java
index f6586081..e94e3328 100644
--- a/app/src/main/java/org/astraea/utils/DataUnit.java
+++ b/app/src/main/java/org/astraea/utils/DataUnit.java
@@ -91,24 +91,6 @@ public enum DataUnit {
     return new DataSize(measurement, this);
   }
 
-  /**
-   * Return a {@link DataSize} based on given measurement and unit.
-   *
-   * <pre>{@code
-   * DataUnit.of(500, DataUnit.KB);   // 500 KB  (500 * 1000 bytes)
-   * DataUnit.of(500, DataUnit.KiB);  // 500 KiB (500 * 1024 bytes)
-   * DataUnit.of(500, DataUnit.Kb);   // 500 Kb  (500 * 1000 bits)
-   * DataUnit.of(500, DataUnit.Kib);  // 500 Kib (500 * 1024 bits)
-   * }</pre>
-   *
-   * @param measurement the data size measurement.
-   * @param unit the data unit of given measurement.
-   * @return a size object of given measurement under specific data unit.
-   */
-  public static DataSize of(long measurement, DataUnit unit) {
-    return new DataSize(measurement, unit);
-  }
-
   /**
    * List of recommended unit for display data, {@link DataSize#toString()} take advantage of this
    * collection to find ideal unit for string formalization.
diff --git a/docker/submit_spark_job.sh b/docker/submit_spark_job.sh
deleted file mode 100755
index f145d250..00000000
--- a/docker/submit_spark_job.sh
+++ /dev/null
@@ -1,154 +0,0 @@
-#!/bin/bash
-
-# =============================[functions]=============================
-
-function showHelp() {
-  echo "Usage: [ENV] submit_spark_job.sh [ ARGUMENTS ]"
-  echo "ENV: "
-  echo "    REPO=astraea/spark-job             set the docker repo"
-  echo "    VERSION=3.1.2                      set version of spark distribution"
-  echo "    DELTA_VERSION=1.0.0                set version of delta distribution"
-  echo "    PY_KAFKA_VERSION=1.7.0             set version of confluent kafka distribution"
-  echo "    RUN=false                          set false if you want to build image only"
-  echo "    MOUNT=host_path:container_path     mount the host path to container path"
-}
-
-# ===============================[checks]===============================
-
-if [[ "$(which docker)" == "" ]]; then
-  echo "you have to install docker"
-  exit 2
-fi
-
-if [[ "$(which ipconfig)" != "" ]]; then
-  address=$(ipconfig getifaddr en0)
-else
-  address=$(hostname -i)
-fi
-
-# =================================[main]=================================
-while [[ $# -gt 0 ]]; do
-  if [[ "$1" == "help" ]]; then
-    showHelp
-    exit 0
-  fi
-  if [[ "$1" != "--"* ]] && [[ "$1" == *".py" ]]; then
-    py_file=$1
-  fi
-  args="$args $1"
-  shift
-done
-
-spark_user=astraea
-version=${REVISION:-${VERSION:-3.1.2}}
-repo=${REPO:-astraea/spark-job}
-image_name="$repo:$version"
-run_container=${RUN:-true}
-ivy_version=2.5.0
-delta_version=${DELTA_VERSION:-1.0.0}
-py_kafka_version=${PY_KAFKA_VERSION:-1.7.0}
-# hardcode version to avoid NPE (see https://issues.apache.org/jira/browse/HADOOP-16410)
-hadoop_version=3.2.2
-docker_dir=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
-dockerfile=$docker_dir/spark-job.dockerfile
-
-echo "# this dockerfile is generated dynamically
-FROM ubuntu:20.04
-
-# Do not ask for confirmations when running apt-get, etc.
-ENV DEBIAN_FRONTEND noninteractive
-
-# install tools
-RUN apt-get update && apt-get upgrade -y && apt-get install -y openjdk-11-jdk wget python3 python3-pip unzip
-
-# add user
-RUN groupadd $spark_user && useradd -ms /bin/bash -g $spark_user $spark_user
-
-# change user
-USER $spark_user
-
-# pre-install python dependencies
-RUN pip3 install confluent-kafka==$py_kafka_version delta-spark==$delta_version pyspark==$version
-
-# install java dependencies
-WORKDIR /tmp
-RUN wget https://dlcdn.apache.org//ant/ivy/${ivy_version}/apache-ivy-${ivy_version}-bin.zip
-RUN unzip apache-ivy-${ivy_version}-bin.zip
-WORKDIR apache-ivy-${ivy_version}
-RUN java -jar ./ivy-${ivy_version}.jar -dependency io.delta delta-core_2.12 $delta_version
-RUN java -jar ./ivy-${ivy_version}.jar -dependency org.apache.spark spark-sql-kafka-0-10_2.12 $version
-RUN java -jar ./ivy-${ivy_version}.jar -dependency org.apache.spark spark-token-provider-kafka-0-10_2.12 $version
-RUN java -jar ./ivy-${ivy_version}.jar -dependency org.apache.hadoop hadoop-azure $hadoop_version
-
-# download spark
-WORKDIR /tmp
-RUN wget https://archive.apache.org/dist/spark/spark-${version}/spark-${version}-bin-hadoop3.2.tgz
-RUN mkdir /home/$spark_user/spark
-RUN tar -zxvf spark-${version}-bin-hadoop3.2.tgz -C /home/$spark_user/spark --strip-components=1
-WORKDIR /home/$spark_user/spark
-" > "$dockerfile"
-
-# build image only if the image does not exist locally
-if [[ "$(docker images -q $image_name 2> /dev/null)" == "" ]]; then
-  docker build -t $image_name -f "$dockerfile" "$docker_dir"
-fi
-
-if [[ "$run_container" != "true" ]]; then
-  echo "docker image: $image_name is created"
-  exit 0
-fi
-
-if [[ "$OSTYPE" == "darwin"* ]]; then
-  echo "This script requires to run container with \"--network host\", but the feature is unsupported by Mac OS"
-  exit 2
-fi
-
-if [[ "$address" == "127.0.0.1" || "$address" == "127.0.1.1" ]]; then
-  echo "the address: Either 127.0.0.1 or 127.0.1.1 can't be used in this script. Please check /etc/hosts"
-  exit 2
-fi
-
-if [[ -z "$py_file" ]]; then
-  echo "failed to get main py file from input arguments"
-  exit 2
-fi
-
-if [[ ! -f "$py_file" ]]; then
-  echo "\"$py_file\" is not a file"
-  exit 2
-fi
-
-# spark stores package files to ~/.ivy2 by default. We keep those files in host path to avoid download them again.
-mkdir -p "$HOME"/.ivy2
-
-py_folder=$(dirname "$py_file")
-
-if [[ -n "$MOUNT" ]]; then
-  index="1"
-  IFS=',' read -ra folders <<< "$MOUNT"
-  for mapping in "${folders[@]}"; do
-    host_folder=$(echo "$mapping" | cut -d : -f 1)
-    container_folder=$(echo "$mapping" | cut -d : -f 2)
-
-    # create folder on host
-    mkdir -p "$host_folder"
-
-    if [[ "$index" == "1" ]]; then
-      hostFolderConfigs="-v $host_folder:$container_folder"
-    else
-      hostFolderConfigs="$hostFolderConfigs -v $host_folder:$container_folder"
-    fi
-    index=$((index+1))
-  done
-
-  docker run \
-    --network host \
-    $hostFolderConfigs \
-    -v "$py_folder":"$py_folder":ro \
-    $image_name ./bin/spark-submit $args
-else
-  docker run \
-    --network host \
-    -v "$py_folder":"$py_folder":ro \
-    $image_name ./bin/spark-submit $args
-fi


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 29c0bbf8..37b42817 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -2,6 +2,7 @@ package org.astraea.performance;
 
 import java.util.Arrays;
 import java.util.List;
+import org.astraea.utils.DataUnit;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -31,7 +32,7 @@ public class ManagerTest {
   void testRandomSize() {
     var argument = new Performance.Argument();
     argument.exeTime = ExeTime.of("3records");
-    argument.recordSize = 102400;
+    argument.recordSize = DataUnit.KiB.of(100);
     var dataManager = new Manager(argument, List.of(), List.of());
     boolean sameSize = dataManager.payload().get().length == dataManager.payload().get().length;
 
@@ -45,7 +46,7 @@ public class ManagerTest {
   void testRandomContent() {
     var argument = new Performance.Argument();
     argument.exeTime = ExeTime.of("2records");
-    argument.recordSize = 102400;
+    argument.recordSize = DataUnit.KiB.of(100);
     var manager = new Manager(argument, List.of(), List.of());
     boolean same = Arrays.equals(manager.payload().get(), manager.payload().get());
 
diff --git a/app/src/test/java/org/astraea/utils/DataSizeTest.java b/app/src/test/java/org/astraea/utils/DataSizeTest.java
index 83d1ca6d..5a21cfdf 100644
--- a/app/src/test/java/org/astraea/utils/DataSizeTest.java
+++ b/app/src/test/java/org/astraea/utils/DataSizeTest.java
@@ -146,6 +146,54 @@ class DataSizeTest {
     assertEquals(new BigInteger(expectedBits), DataUnit.valueOf(unitName).of(1).bits());
   }
 
+  @Test
+  void parseDataSize() {
+    var converter = new DataSize.Converter();
+    assertEquals(DataUnit.Bit.of(100).toString(), converter.convert("100Bit").toString());
+    assertEquals(DataUnit.Kb.of(100).toString(), converter.convert("100 Kb").toString());
+    assertEquals(DataUnit.Mb.of(100).toString(), converter.convert("100 Mb").toString());
+    assertEquals(DataUnit.Gb.of(100).toString(), converter.convert("100 Gb").toString());
+    assertEquals(DataUnit.Tb.of(100).toString(), converter.convert("100 Tb").toString());
+    assertEquals(DataUnit.Pb.of(100).toString(), converter.convert("100 Pb").toString());
+    assertEquals(DataUnit.Eb.of(100).toString(), converter.convert("100 Eb").toString());
+    assertEquals(DataUnit.Zb.of(100).toString(), converter.convert("100 Zb").toString());
+    assertEquals(DataUnit.Yb.of(100).toString(), converter.convert("100 Yb").toString());
+
+    assertEquals(DataUnit.Kib.of(100).toString(), converter.convert("100 Kib").toString());
+    assertEquals(DataUnit.Mib.of(100).toString(), converter.convert("100 Mib").toString());
+    assertEquals(DataUnit.Gib.of(100).toString(), converter.convert("100 Gib").toString());
+    assertEquals(DataUnit.Tib.of(100).toString(), converter.convert("100 Tib").toString());
+    assertEquals(DataUnit.Pib.of(100).toString(), converter.convert("100 Pib").toString());
+    assertEquals(DataUnit.Eib.of(100).toString(), converter.convert("100 Eib").toString());
+    assertEquals(DataUnit.Zib.of(100).toString(), converter.convert("100 Zib").toString());
+    assertEquals(DataUnit.Yib.of(100).toString(), converter.convert("100 Yib").toString());
+
+    assertEquals(DataUnit.Byte.of(100).toString(), converter.convert("100Byte").toString());
+    assertEquals(DataUnit.KB.of(100).toString(), converter.convert("100 KB").toString());
+    assertEquals(DataUnit.MB.of(100).toString(), converter.convert("100 MB").toString());
+    assertEquals(DataUnit.GB.of(100).toString(), converter.convert("100 GB").toString());
+    assertEquals(DataUnit.TB.of(100).toString(), converter.convert("100 TB").toString());
+    assertEquals(DataUnit.PB.of(100).toString(), converter.convert("100 PB").toString());
+    assertEquals(DataUnit.EB.of(100).toString(), converter.convert("100 EB").toString());
+    assertEquals(DataUnit.ZB.of(100).toString(), converter.convert("100 ZB").toString());
+    assertEquals(DataUnit.YB.of(100).toString(), converter.convert("100 YB").toString());
+
+    assertEquals(DataUnit.KiB.of(100).toString(), converter.convert("100 KiB").toString());
+    assertEquals(DataUnit.MiB.of(100).toString(), converter.convert("100 MiB").toString());
+    assertEquals(DataUnit.GiB.of(100).toString(), converter.convert("100 GiB").toString());
+    assertEquals(DataUnit.TiB.of(100).toString(), converter.convert("100 TiB").toString());
+    assertEquals(DataUnit.PiB.of(100).toString(), converter.convert("100 PiB").toString());
+    assertEquals(DataUnit.EiB.of(100).toString(), converter.convert("100 EiB").toString());
+    assertEquals(DataUnit.ZiB.of(100).toString(), converter.convert("100 ZiB").toString());
+    assertEquals(DataUnit.YiB.of(100).toString(), converter.convert("100 YiB").toString());
+
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("5000 MB xxx"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("5000 MB per second"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("xxx 5000 MB"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("5000 MB 400GB"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("6.00 MB"));
+  }
+
   @Test
   void measurement() {
     var value = DataUnit.PB.of(1);
@@ -227,7 +275,7 @@ class DataSizeTest {
 
   @Test
   void bits() {
-    BigInteger bit5566 = DataUnit.of(5566, DataUnit.Bit).bits();
+    BigInteger bit5566 = DataUnit.Bit.of(5566).bits();
 
     assertEquals(BigInteger.valueOf(5566), bit5566);
   }