Introduce to integrate converter and validator

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

@@ -0,0 +1,74 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.UnixStyleUsageFormatter;
+import java.io.*;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.CommonClientConfigs;
+
+/** This basic argument defines the common property used by all kafka clients. */
+public abstract class Argument {
+
+  /**
+   * Side effect: parse args into toolArgument
+   *
+   * @param toolArgument An argument object that the user want.
+   * @param args Command line arguments that are put into main function.
+   */
+  public static <T> T parse(T toolArgument, String[] args) {
+    JCommander jc = JCommander.newBuilder().addObject(toolArgument).build();
+    jc.setUsageFormatter(new UnixStyleUsageFormatter(jc));
+    try {
+      jc.parse(args);
+    } catch (ParameterException pe) {
+      var sb = new StringBuilder();
+      jc.getUsageFormatter().usage(sb);
+      throw new ParameterException(pe.getMessage() + "\n" + sb);
+    }
+    return toolArgument;
+  }
+
+  @Parameter(
+      names = {"--bootstrap.servers"},
+      description = "String: server to connect to",
+      validateWith = NonEmptyStringField.class,
+      required = true)
+  public String brokers;
+
+  /**
+   * @param propsFile file path containing the properties to be passed to kafka
+   * @return the kafka props consists of bootstrap servers and all props from file (if it is
+   *     existent)
+   */
+  Map<String, Object> properties(String propsFile) {
+    var props = new Properties();
+    if (propsFile != null) {
+      try (var input = new FileInputStream(propsFile)) {
+        props.load(input);
+      } catch (IOException e) {
+        throw new UncheckedIOException(e);
+      }
+    }
+    props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokers);
+    return props.entrySet().stream()
+        .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
+  }
+
+  @Parameter(
+      names = {"--prop.file"},
+      description = "the file path containing the properties to be passed to kafka admin",
+      validateWith = NonEmptyStringField.class)
+  public String propFile;
+
+  /**
+   * @return the kafka props consists of bootstrap servers and all props from file (if it is
+   *     existent)
+   */
+  public Map<String, Object> props() {
+    return properties(propFile);
+  }
+}

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

@@ -0,0 +1,17 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+
+public class BooleanField implements NonEmptyField<Boolean> {
+  @Override
+  public Boolean convert(String value) {
+    return Boolean.parseBoolean(value);
+  }
+
+  @Override
+  public void validate(String name, String value) throws ParameterException {
+    NonEmptyField.super.validate(name, value);
+    if (!("false".equalsIgnoreCase(value) || "true".equalsIgnoreCase(value)))
+      throw new ParameterException(value + " is not boolean type");
+  }
+}

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

@@ -0,0 +1,34 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.kafka.common.record.CompressionType;
+
+public class CompressionField implements NonEmptyField<CompressionType> {
+  /**
+   * @param value Name of compression type. Accept lower-case name only ("none", "gzip", "snappy",
+   *     "lz4", "zstd").
+   */
+  @Override
+  public CompressionType convert(String value) {
+    try {
+      // `CompressionType#forName` accept lower-case name only.
+      return CompressionType.forName(value);
+    } catch (IllegalArgumentException e) {
+      throw new ParameterException(
+          "the "
+              + value
+              + " is unsupported. The supported algorithms are "
+              + Stream.of(CompressionType.values())
+                  .map(CompressionType::name)
+                  .collect(Collectors.joining(",")));
+    }
+  }
+
+  @Override
+  public void validate(String name, String value) throws ParameterException {
+    NonEmptyField.super.validate(name, value);
+    convert(value);
+  }
+}

app/src/main/java/org/astraea/argument/converter/DurationConverter.java → app/src/main/java/org/astraea/argument/DurationField.java

@@ -1,7 +1,5 @@
-package org.astraea.argument.converter;
+package org.astraea.argument;
 
-import com.beust.jcommander.IParameterValidator;
-import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.ParameterException;
 import java.time.Duration;
 import java.util.regex.Matcher;
@@ -42,7 +40,7 @@
  *   <li><b>(doesn't work)</b> {@code "0.5" to {@code Duration.ofMillis(500)}}
  * </ul>
  */
-public class DurationConverter implements IStringConverter<Duration>, IParameterValidator {
+public class DurationField implements Field<Duration> {
 
   static final Pattern TIME_PATTERN =
       Pattern.compile("^(?<value>[0-9]+)(?<unit>days|day|h|m|s|ms|us|ns|)$");

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

@@ -0,0 +1,21 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.IParameterValidator;
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.ParameterException;
+
+public interface Field<T> extends IStringConverter<T>, IParameterValidator {
+
+  /** @return an object of type T created from the parameter value. */
+  @Override
+  T convert(String value);
+
+  /**
+   * Validate the parameter.
+   *
+   * @param name The name of the parameter (e.g. "-host").
+   * @param value The value of the parameter that we need to validate
+   * @throws ParameterException Thrown if the value of the parameter is invalid.
+   */
+  void validate(String name, String value) throws ParameterException;
+}

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

@@ -0,0 +1,12 @@
+package org.astraea.argument;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class IntegerSetField implements SetField<Integer> {
+  @Override
+  public Set<Integer> convert(String value) {
+    return Stream.of(value.split(SEPARATOR)).map(Integer::valueOf).collect(Collectors.toSet());
+  }
+}

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

@@ -0,0 +1,12 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+
+public interface NonEmptyField<T> extends Field<T> {
+
+  @Override
+  default void validate(String name, String value) throws ParameterException {
+    if (value == null || value.isBlank())
+      throw new ParameterException(name + " should not be empty.");
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class NonEmptyStringField implements NonEmptyField<String> {
+  @Override
+  public String convert(String value) {
+    return value;
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class NonNegativeDoubleField implements PositiveNumberField<Double> {
+  @Override
+  public Double convert(String value) {
+    return Double.parseDouble(value);
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class NonNegativeIntegerField implements PositiveNumberField<Integer> {
+  @Override
+  public Integer convert(String value) {
+    return Integer.parseInt(value);
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class NonNegativeLongField implements PositiveNumberField<Long> {
+  @Override
+  public Long convert(String value) {
+    return Long.parseLong(value);
+  }
+}

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

@@ -0,0 +1,12 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+
+public interface NonNegativeNumberField<T extends Number> extends NonEmptyField<T> {
+
+  @Override
+  default void validate(String name, String value) throws ParameterException {
+    NonEmptyField.super.validate(name, value);
+    if (Long.parseLong(value) < 0) throw new ParameterException(name + " should not be negative.");
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class NonNegativeShortField implements PositiveNumberField<Short> {
+  @Override
+  public Short convert(String value) {
+    return Short.parseShort(value);
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class PositiveDoubleField implements PositiveNumberField<Double> {
+  @Override
+  public Double convert(String value) {
+    return Double.parseDouble(value);
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class PositiveIntegerField implements PositiveNumberField<Integer> {
+  @Override
+  public Integer convert(String value) {
+    return Integer.parseInt(value);
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class PositiveLongField implements PositiveNumberField<Long> {
+  @Override
+  public Long convert(String value) {
+    return Long.parseLong(value);
+  }
+}

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

@@ -0,0 +1,11 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+
+public interface PositiveNumberField<T extends Number> extends Field<T> {
+
+  @Override
+  default void validate(String name, String value) throws ParameterException {
+    if (Long.parseLong(value) <= 0) throw new ParameterException(name + " should be positive.");
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.astraea.argument;
+
+public class PositiveShortField implements PositiveNumberField<Short> {
+  @Override
+  public Short convert(String value) {
+    return Short.parseShort(value);
+  }
+}