Repository: skiptests/astraea
Pull request #247 (2022-02-22 05:59:41)

Fix flaky node exporter check
```
TASK [Launch Node Exporter on all host] ***********************************************************************************************************************************************************************************************************************************************************************************************************************************************
skipping: [127.0.0.1]
fatal: [parrotparty1]: FAILED! => {"changed": true, "cmd": ["/home/garyparrot/ansible-automation/astraea/docker/start_node_exporter.sh"], "delta": "0:00:00.381411", "end": "2022-02-22 05:41:13.060227", "msg": "non-zero return code", "rc": 3, "start": "2022-02-22 05:41:12.678816", "stderr": "", "stderr_lines": [], "stdout": "[INFO] Container ID of node_exporter: 8a6fe4f2ac6ed951dc0e6f16fc64d77ff9fb053170e901fdc857e51f6430fa77\n[ERROR] Cannot access node_exporter metric server: http://0.0.0.0:11105. Something go wrong!\n[ERROR] Execute \"docker logs 8a6fe4f2ac6ed951dc0e6f16fc64d77ff9fb053170e901fdc857e51f6430fa77\" to see what's going on.", "stdout_lines": ["[INFO] Container ID of node_exporter: 8a6fe4f2ac6ed951dc0e6f16fc64d77ff9fb053170e901fdc857e51f6430fa77", "[ERROR] Cannot access node_exporter metric server: http://0.0.0.0:11105. Something go wrong!", "[ERROR] Execute \"docker logs 8a6fe4f2ac6ed951dc0e6f16fc64d77ff9fb053170e901fdc857e51f6430fa77\" to see what's going on."]}
fatal: [parrotparty2]: FAILED! => {"changed": true, "cmd": ["/home/garyparrot/ansible-automation/astraea/docker/start_node_exporter.sh"], "delta": "0:00:00.422028", "end": "2022-02-22 05:41:13.141300", "msg": "non-zero return code", "rc": 3, "start": "2022-02-22 05:41:12.719272", "stderr": "", "stderr_lines": [], "stdout": "[INFO] Container ID of node_exporter: 988c7fad82d3a5bda3471f7207419cb121ad6160854022edfe4aaf86d2fcbf96\n[ERROR] Cannot access node_exporter metric server: http://0.0.0.0:11105. Something go wrong!\n[ERROR] Execute \"docker logs 988c7fad82d3a5bda3471f7207419cb121ad6160854022edfe4aaf86d2fcbf96\" to see what's going on.", "stdout_lines": ["[INFO] Container ID of node_exporter: 988c7fad82d3a5bda3471f7207419cb121ad6160854022edfe4aaf86d2fcbf96", "[ERROR] Cannot access node_exporter metric server: http://0.0.0.0:11105. Something go wrong!", "[ERROR] Execute \"docker logs 988c7fad82d3a5bda3471f7207419cb121ad6160854022edfe4aaf86d2fcbf96\" to see what's going on."]}
fatal: [parrotparty3]: FAILED! => {"changed": true, "cmd": ["/home/garyparrot/ansible-automation/astraea/docker/start_node_exporter.sh"], "delta": "0:00:00.495669", "end": "2022-02-22 05:41:13.219281", "msg": "non-zero return code", "rc": 3, "start": "2022-02-22 05:41:12.723612", "stderr": "", "stderr_lines": [], "stdout": "[INFO] Container ID of node_exporter: afc802c9e4e58a3bcc863692fd4133832fc2f6d884c026ff4968b981ecb512b3\n[ERROR] Cannot access node_exporter metric server: http://0.0.0.0:11105. Something go wrong!\n[ERROR] Execute \"docker logs afc802c9e4e58a3bcc863692fd4133832fc2f6d884c026ff4968b981ecb512b3\" to see what's going on.", "stdout_lines": ["[INFO] Container ID of node_exporter: afc802c9e4e58a3bcc863692fd4133832fc2f6d884c026ff4968b981ecb512b3", "[ERROR] Cannot access node_exporter metric server: http://0.0.0.0:11105. Something go wrong!", "[ERROR] Execute \"docker logs afc802c9e4e58a3bcc863692fd4133832fc2f6d884c026ff4968b981ecb512b3\" to see what's going on."]}
```

`start_node_exporter.sh` 腳本最後會檢查 server 有沒有跑起來，不過這個檢查和 server 是否真的起來了是一個 race condition. 。在艱困的實驗環境中這個檢查一直報錯。

這個 PR 把這個檢查移除

chia7712: > start_node_exporter.sh 腳本最後會檢查 server 有沒有跑起來，不過這個檢查和 server 是否真的起來了是一個 race condition. 。在艱困的實驗環境中這個檢查一直報錯。

請問一下為何會跑不起來？confluent的啟動時間比較久嗎？
chia7712: @garyparrot 麻煩rebase一下程式碼，看看是否還會有flaky

Patch:
diff --git a/README.md b/README.md
index 20d5a504..38102744 100644
--- a/README.md
+++ b/README.md
@@ -112,7 +112,7 @@ For example. Assume you have two Kafka brokers,
 You can execute the following command to create a Prometheus instance that fetches data from the above 4 exporters.
 
 ```shell
-./docker/start_prometheus.sh start 192.168.0.1:10558,192.168.0.2:10558 192.168.0.1:9100,192.168.0.2:9100
+./docker/start_prometheus.sh start
 ```
 
 The console will show the http address of prometheus service, also some hints for you to set up Grafana. See next [section](#run-grafana) for
@@ -129,6 +129,16 @@ prometheus-9090
 [INFO] =================================================
 ```
 
+There are two ways to change the prometheus configuration.
+
+```shell
+# refresh prometheus config by command-line arguments
+./docker/start_prometheus.sh refresh host1:1111,host2:1111 host1:2222,host2:2222
+
+# refresh prometheus config by specific file
+./docker/start_prometheus.sh refresh ./path/to/config/file/prometheus.yml
+```
+
 #### Update Prometheus configuration
 
 To update your Prometheus configuration, there are two ways.
diff --git a/app/src/main/java/org/astraea/argument/BooleanField.java b/app/src/main/java/org/astraea/argument/BooleanField.java
index 86165c35..51ea1b82 100644
--- a/app/src/main/java/org/astraea/argument/BooleanField.java
+++ b/app/src/main/java/org/astraea/argument/BooleanField.java
@@ -2,15 +2,14 @@ package org.astraea.argument;
 
 import com.beust.jcommander.ParameterException;
 
-public class BooleanField implements NonEmptyField<Boolean> {
+public class BooleanField extends Field<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);
+  protected void check(String name, String value) throws ParameterException {
     if (!("false".equalsIgnoreCase(value) || "true".equalsIgnoreCase(value)))
       throw new ParameterException(value + " is not boolean type");
   }
diff --git a/app/src/main/java/org/astraea/argument/CompressionField.java b/app/src/main/java/org/astraea/argument/CompressionField.java
index f5df778e..4e937574 100644
--- a/app/src/main/java/org/astraea/argument/CompressionField.java
+++ b/app/src/main/java/org/astraea/argument/CompressionField.java
@@ -5,7 +5,7 @@ import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.kafka.common.record.CompressionType;
 
-public class CompressionField implements NonEmptyField<CompressionType> {
+public class CompressionField extends Field<CompressionType> {
   /**
    * @param value Name of compression type. Accept lower-case name only ("none", "gzip", "snappy",
    *     "lz4", "zstd").
@@ -27,8 +27,7 @@ public class CompressionField implements NonEmptyField<CompressionType> {
   }
 
   @Override
-  public void validate(String name, String value) throws ParameterException {
-    NonEmptyField.super.validate(name, value);
+  protected void check(String name, String value) throws ParameterException {
     convert(value);
   }
 }
diff --git a/app/src/main/java/org/astraea/argument/DurationField.java b/app/src/main/java/org/astraea/argument/DurationField.java
index 6049a478..f42303d8 100644
--- a/app/src/main/java/org/astraea/argument/DurationField.java
+++ b/app/src/main/java/org/astraea/argument/DurationField.java
@@ -40,7 +40,7 @@ import java.util.regex.Pattern;
  *   <li><b>(doesn't work)</b> {@code "0.5" to {@code Duration.ofMillis(500)}}
  * </ul>
  */
-public class DurationField implements Field<Duration> {
+public class DurationField extends Field<Duration> {
 
   static final Pattern TIME_PATTERN =
       Pattern.compile("^(?<value>[0-9]+)(?<unit>days|day|h|m|s|ms|us|ns|)$");
@@ -75,7 +75,7 @@ public class DurationField implements Field<Duration> {
   }
 
   @Override
-  public void validate(String name, String value) throws ParameterException {
+  protected void check(String name, String value) throws ParameterException {
     if (!TIME_PATTERN.matcher(value).find())
       throw new ParameterException(
           "field \"" + name + "\"'s value \"" + value + "\" doesn't match time format");
diff --git a/app/src/main/java/org/astraea/argument/Field.java b/app/src/main/java/org/astraea/argument/Field.java
index 1b281594..15122d2f 100644
--- a/app/src/main/java/org/astraea/argument/Field.java
+++ b/app/src/main/java/org/astraea/argument/Field.java
@@ -4,11 +4,22 @@ import com.beust.jcommander.IParameterValidator;
 import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.ParameterException;
 
-public interface Field<T> extends IStringConverter<T>, IParameterValidator {
+public abstract class Field<T> implements IStringConverter<T>, IParameterValidator {
 
   /** @return an object of type T created from the parameter value. */
   @Override
-  T convert(String value);
+  public abstract T convert(String value);
+
+  /**
+   * check the parameter before converting
+   *
+   * @param name The name of the parameter (e.g. "-host").
+   * @param value The value of the parameter that we need to validate. never null or empty
+   * @throws ParameterException Thrown if the value of the parameter is invalid.
+   */
+  protected void check(String name, String value) throws ParameterException {
+    // do nothing
+  }
 
   /**
    * Validate the parameter.
@@ -17,5 +28,9 @@ public interface Field<T> extends IStringConverter<T>, IParameterValidator {
    * @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;
+  public final void validate(String name, String value) throws ParameterException {
+    if (value == null || value.isBlank())
+      throw new ParameterException(name + " should not be empty.");
+    check(name, value);
+  }
 }
diff --git a/app/src/main/java/org/astraea/argument/IntegerSetField.java b/app/src/main/java/org/astraea/argument/IntegerSetField.java
index 8344d81c..7ed2e703 100644
--- a/app/src/main/java/org/astraea/argument/IntegerSetField.java
+++ b/app/src/main/java/org/astraea/argument/IntegerSetField.java
@@ -4,7 +4,7 @@ import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-public class IntegerSetField implements SetField<Integer> {
+public class IntegerSetField extends SetField<Integer> {
   @Override
   public Set<Integer> convert(String value) {
     return Stream.of(value.split(SEPARATOR)).map(Integer::valueOf).collect(Collectors.toSet());
diff --git a/app/src/main/java/org/astraea/argument/NonEmptyField.java b/app/src/main/java/org/astraea/argument/NonEmptyField.java
deleted file mode 100644
index 4c89054d..00000000
--- a/app/src/main/java/org/astraea/argument/NonEmptyField.java
+++ /dev/null
@@ -1,12 +0,0 @@
-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.");
-  }
-}
diff --git a/app/src/main/java/org/astraea/argument/NonEmptyStringField.java b/app/src/main/java/org/astraea/argument/NonEmptyStringField.java
index 0837b85d..3edafd10 100644
--- a/app/src/main/java/org/astraea/argument/NonEmptyStringField.java
+++ b/app/src/main/java/org/astraea/argument/NonEmptyStringField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class NonEmptyStringField implements NonEmptyField<String> {
+public class NonEmptyStringField extends Field<String> {
   @Override
   public String convert(String value) {
     return value;
diff --git a/app/src/main/java/org/astraea/argument/NonNegativeDoubleField.java b/app/src/main/java/org/astraea/argument/NonNegativeDoubleField.java
index f063a4f9..44617641 100644
--- a/app/src/main/java/org/astraea/argument/NonNegativeDoubleField.java
+++ b/app/src/main/java/org/astraea/argument/NonNegativeDoubleField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class NonNegativeDoubleField implements PositiveNumberField<Double> {
+public class NonNegativeDoubleField extends NonNegativeNumberField<Double> {
   @Override
   public Double convert(String value) {
     return Double.parseDouble(value);
diff --git a/app/src/main/java/org/astraea/argument/NonNegativeIntegerField.java b/app/src/main/java/org/astraea/argument/NonNegativeIntegerField.java
index 2b6b87ac..05c6ac6d 100644
--- a/app/src/main/java/org/astraea/argument/NonNegativeIntegerField.java
+++ b/app/src/main/java/org/astraea/argument/NonNegativeIntegerField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class NonNegativeIntegerField implements PositiveNumberField<Integer> {
+public class NonNegativeIntegerField extends NonNegativeNumberField<Integer> {
   @Override
   public Integer convert(String value) {
     return Integer.parseInt(value);
diff --git a/app/src/main/java/org/astraea/argument/NonNegativeLongField.java b/app/src/main/java/org/astraea/argument/NonNegativeLongField.java
index ff4f7566..1baaf119 100644
--- a/app/src/main/java/org/astraea/argument/NonNegativeLongField.java
+++ b/app/src/main/java/org/astraea/argument/NonNegativeLongField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class NonNegativeLongField implements PositiveNumberField<Long> {
+public class NonNegativeLongField extends NonNegativeNumberField<Long> {
   @Override
   public Long convert(String value) {
     return Long.parseLong(value);
diff --git a/app/src/main/java/org/astraea/argument/NonNegativeNumberField.java b/app/src/main/java/org/astraea/argument/NonNegativeNumberField.java
index 84da9188..dc9e0a6d 100644
--- a/app/src/main/java/org/astraea/argument/NonNegativeNumberField.java
+++ b/app/src/main/java/org/astraea/argument/NonNegativeNumberField.java
@@ -2,11 +2,10 @@ package org.astraea.argument;
 
 import com.beust.jcommander.ParameterException;
 
-public interface NonNegativeNumberField<T extends Number> extends NonEmptyField<T> {
+public abstract class NonNegativeNumberField<T extends Number> extends Field<T> {
 
   @Override
-  default void validate(String name, String value) throws ParameterException {
-    NonEmptyField.super.validate(name, value);
+  protected void check(String name, String value) throws ParameterException {
     if (Long.parseLong(value) < 0) throw new ParameterException(name + " should not be negative.");
   }
 }
diff --git a/app/src/main/java/org/astraea/argument/NonNegativeShortField.java b/app/src/main/java/org/astraea/argument/NonNegativeShortField.java
index 57ea3273..a8032303 100644
--- a/app/src/main/java/org/astraea/argument/NonNegativeShortField.java
+++ b/app/src/main/java/org/astraea/argument/NonNegativeShortField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class NonNegativeShortField implements PositiveNumberField<Short> {
+public class NonNegativeShortField extends NonNegativeNumberField<Short> {
   @Override
   public Short convert(String value) {
     return Short.parseShort(value);
diff --git a/app/src/main/java/org/astraea/argument/PositiveDoubleField.java b/app/src/main/java/org/astraea/argument/PositiveDoubleField.java
index 0a97144d..b06c2686 100644
--- a/app/src/main/java/org/astraea/argument/PositiveDoubleField.java
+++ b/app/src/main/java/org/astraea/argument/PositiveDoubleField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class PositiveDoubleField implements PositiveNumberField<Double> {
+public class PositiveDoubleField extends PositiveNumberField<Double> {
   @Override
   public Double convert(String value) {
     return Double.parseDouble(value);
diff --git a/app/src/main/java/org/astraea/argument/PositiveIntegerField.java b/app/src/main/java/org/astraea/argument/PositiveIntegerField.java
index 38669743..8d4ed69e 100644
--- a/app/src/main/java/org/astraea/argument/PositiveIntegerField.java
+++ b/app/src/main/java/org/astraea/argument/PositiveIntegerField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class PositiveIntegerField implements PositiveNumberField<Integer> {
+public class PositiveIntegerField extends PositiveNumberField<Integer> {
   @Override
   public Integer convert(String value) {
     return Integer.parseInt(value);
diff --git a/app/src/main/java/org/astraea/argument/PositiveLongField.java b/app/src/main/java/org/astraea/argument/PositiveLongField.java
index 79d5cdcd..58aa3c40 100644
--- a/app/src/main/java/org/astraea/argument/PositiveLongField.java
+++ b/app/src/main/java/org/astraea/argument/PositiveLongField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class PositiveLongField implements PositiveNumberField<Long> {
+public class PositiveLongField extends PositiveNumberField<Long> {
   @Override
   public Long convert(String value) {
     return Long.parseLong(value);
diff --git a/app/src/main/java/org/astraea/argument/PositiveNumberField.java b/app/src/main/java/org/astraea/argument/PositiveNumberField.java
index d6ac1c1c..636f2307 100644
--- a/app/src/main/java/org/astraea/argument/PositiveNumberField.java
+++ b/app/src/main/java/org/astraea/argument/PositiveNumberField.java
@@ -2,10 +2,10 @@ package org.astraea.argument;
 
 import com.beust.jcommander.ParameterException;
 
-public interface PositiveNumberField<T extends Number> extends Field<T> {
+public abstract class PositiveNumberField<T extends Number> extends Field<T> {
 
   @Override
-  default void validate(String name, String value) throws ParameterException {
+  protected void check(String name, String value) throws ParameterException {
     if (Long.parseLong(value) <= 0) throw new ParameterException(name + " should be positive.");
   }
 }
diff --git a/app/src/main/java/org/astraea/argument/PositiveShortField.java b/app/src/main/java/org/astraea/argument/PositiveShortField.java
index e9f04ba8..32ace72c 100644
--- a/app/src/main/java/org/astraea/argument/PositiveShortField.java
+++ b/app/src/main/java/org/astraea/argument/PositiveShortField.java
@@ -1,6 +1,6 @@
 package org.astraea.argument;
 
-public class PositiveShortField implements PositiveNumberField<Short> {
+public class PositiveShortField extends PositiveNumberField<Short> {
   @Override
   public Short convert(String value) {
     return Short.parseShort(value);
diff --git a/app/src/main/java/org/astraea/argument/SetField.java b/app/src/main/java/org/astraea/argument/SetField.java
index e543ff75..387df36a 100644
--- a/app/src/main/java/org/astraea/argument/SetField.java
+++ b/app/src/main/java/org/astraea/argument/SetField.java
@@ -3,11 +3,11 @@ package org.astraea.argument;
 import com.beust.jcommander.ParameterException;
 import java.util.Set;
 
-public interface SetField<T> extends Field<Set<T>> {
-  String SEPARATOR = ",";
+public abstract class SetField<T> extends Field<Set<T>> {
+  protected static final String SEPARATOR = ",";
 
   @Override
-  default void validate(String name, String value) throws ParameterException {
+  protected void check(String name, String value) throws ParameterException {
     if (value == null || value.isBlank() || Set.of(value.split(SEPARATOR)).isEmpty())
       throw new ParameterException("set type can't be empty");
   }
diff --git a/app/src/main/java/org/astraea/argument/StringMapField.java b/app/src/main/java/org/astraea/argument/StringMapField.java
new file mode 100644
index 00000000..e54b3d08
--- /dev/null
+++ b/app/src/main/java/org/astraea/argument/StringMapField.java
@@ -0,0 +1,27 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class StringMapField extends Field<Map<String, String>> {
+
+  @Override
+  public Map<String, String> convert(String value) {
+    return Arrays.stream(value.split(","))
+        .map(
+            item -> {
+              var keyValue = item.split("=");
+              if (keyValue.length != 2) throw new ParameterException("incorrect format: " + item);
+              return Map.entry(keyValue[0], keyValue[1]);
+            })
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  @Override
+  protected void check(String name, String value) throws ParameterException {
+    if (!value.contains("=") || convert(value).isEmpty())
+      throw new ParameterException("incorrect format: " + value);
+  }
+}
diff --git a/app/src/main/java/org/astraea/argument/StringSetField.java b/app/src/main/java/org/astraea/argument/StringSetField.java
index 5f74f694..bb264ff0 100644
--- a/app/src/main/java/org/astraea/argument/StringSetField.java
+++ b/app/src/main/java/org/astraea/argument/StringSetField.java
@@ -2,7 +2,7 @@ package org.astraea.argument;
 
 import java.util.Set;
 
-public class StringSetField implements SetField<String> {
+public class StringSetField extends SetField<String> {
 
   @Override
   public Set<String> convert(String value) {
diff --git a/app/src/main/java/org/astraea/automation/Automation.java b/app/src/main/java/org/astraea/automation/Automation.java
index 8a0679d3..7486c675 100644
--- a/app/src/main/java/org/astraea/automation/Automation.java
+++ b/app/src/main/java/org/astraea/automation/Automation.java
@@ -54,15 +54,14 @@ public class Automation {
       else times = Integer.parseInt(properties.getProperty("--time"));
 
       while (i < times) {
-        var str =
+        var result =
             Performance.execute(
                 org.astraea.argument.Argument.parse(
                     new Performance.Argument(), performanceArgs(properties)));
         i++;
         if (whetherDeleteTopic) {
           try (final AdminClient adminClient = KafkaAdminClient.create(config)) {
-            var topicName = str.get();
-            adminClient.deleteTopics(List.of(topicName));
+            adminClient.deleteTopics(List.of(result.topicName()));
           }
         }
         System.out.println("=============== " + i + " time Performance Complete! ===============");
diff --git a/app/src/main/java/org/astraea/concurrent/Executor.java b/app/src/main/java/org/astraea/concurrent/Executor.java
new file mode 100644
index 00000000..624b21c4
--- /dev/null
+++ b/app/src/main/java/org/astraea/concurrent/Executor.java
@@ -0,0 +1,21 @@
+package org.astraea.concurrent;
+
+@FunctionalInterface
+public interface Executor extends AutoCloseable {
+
+  /**
+   * @return the state of this executor
+   * @throws InterruptedException This is an expected exception if your executor needs to call
+   *     blocking method. This exception is not printed to console.
+   */
+  State execute() throws InterruptedException;
+
+  /** close this executor. */
+  @Override
+  default void close() {}
+
+  /**
+   * If this executor is in blocking mode, this method offers a way to wake up executor to close.
+   */
+  default void wakeup() {}
+}
diff --git a/app/src/main/java/org/astraea/concurrent/State.java b/app/src/main/java/org/astraea/concurrent/State.java
new file mode 100644
index 00000000..1a55800f
--- /dev/null
+++ b/app/src/main/java/org/astraea/concurrent/State.java
@@ -0,0 +1,8 @@
+package org.astraea.concurrent;
+
+public enum State {
+  /** this thread is done. ThreadPool will call close to release the thread. */
+  DONE,
+  /** this thread is running */
+  RUNNING
+}
diff --git a/app/src/main/java/org/astraea/concurrent/ThreadPool.java b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
index d96263f6..ec32bfcb 100644
--- a/app/src/main/java/org/astraea/concurrent/ThreadPool.java
+++ b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
@@ -30,48 +30,11 @@ public interface ThreadPool extends AutoCloseable {
     return new Builder();
   }
 
-  @FunctionalInterface
-  interface Executor extends AutoCloseable {
-
-    enum State {
-      DONE,
-      RUNNING
-    }
-
-    /**
-     * @return the state of this executor
-     * @throws InterruptedException This is an expected exception if your executor needs to call
-     *     blocking method. This exception is not printed to console.
-     */
-    State execute() throws InterruptedException;
-
-    /** close this executor. */
-    default void close() {}
-
-    /**
-     * If this executor is in blocking mode, this method offers a way to wake up executor to close.
-     */
-    default void wakeup() {}
-  }
-
   class Builder {
     private final List<Executor> executors = new ArrayList<>();
 
     private Builder() {}
 
-    public Builder runnable(Runnable runnable) {
-      return executor(
-          () -> {
-            runnable.run();
-            return Executor.State.RUNNING;
-          });
-    }
-
-    public Builder runnables(Collection<Runnable> runnables) {
-      runnables.forEach(this::runnable);
-      return this;
-    }
-
     public Builder executor(Executor executor) {
       return executors(List.of(executor));
     }
@@ -91,7 +54,7 @@ public interface ThreadPool extends AutoCloseable {
                   () -> {
                     try {
                       while (!closed.get()) {
-                        if (executor.execute() == Executor.State.DONE) break;
+                        if (executor.execute() == State.DONE) break;
                       }
                     } catch (InterruptedException e) {
                       // swallow
diff --git a/app/src/main/java/org/astraea/metrics/MetricExplorer.java b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
index 1ba15b0e..437ecd56 100644
--- a/app/src/main/java/org/astraea/metrics/MetricExplorer.java
+++ b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
@@ -16,7 +16,6 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import javax.management.openmbean.CompositeDataSupport;
 import javax.management.remote.JMXServiceURL;
-import org.astraea.argument.NonEmptyField;
 import org.astraea.metrics.jmx.BeanObject;
 import org.astraea.metrics.jmx.BeanQuery;
 import org.astraea.metrics.jmx.MBeanClient;
@@ -206,7 +205,7 @@ public class MetricExplorer {
         description = "Show the list view of MBeans' domain name & properties")
     boolean viewObjectNameList = false;
 
-    public static class JmxServerUrlField implements NonEmptyField<JMXServiceURL> {
+    public static class JmxServerUrlField extends org.astraea.argument.Field<JMXServiceURL> {
 
       /** This regex used to test if a string look like a JMX URL */
       static Pattern patternOfJmxUrlStart = Pattern.compile("^service:jmx:rmi://");
diff --git a/app/src/main/java/org/astraea/partitioner/NodeId.java b/app/src/main/java/org/astraea/partitioner/NodeId.java
deleted file mode 100644
index caa0fc53..00000000
--- a/app/src/main/java/org/astraea/partitioner/NodeId.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package org.astraea.partitioner;
-
-/** We use this interface to replace the integer for more readable code. */
-@FunctionalInterface
-public interface NodeId extends Comparable<NodeInfo> {
-
-  static NodeId of(int id) {
-    return () -> id;
-  }
-
-  /** @return The id for this node */
-  int id();
-
-  @Override
-  default int compareTo(NodeInfo o) {
-    return Integer.compare(id(), o.id());
-  }
-}
diff --git a/app/src/main/java/org/astraea/partitioner/NodeInfo.java b/app/src/main/java/org/astraea/partitioner/NodeInfo.java
index 9a54c41f..f0f8029a 100644
--- a/app/src/main/java/org/astraea/partitioner/NodeInfo.java
+++ b/app/src/main/java/org/astraea/partitioner/NodeInfo.java
@@ -2,7 +2,7 @@ package org.astraea.partitioner;
 
 import java.util.Objects;
 
-public interface NodeInfo extends Comparable<NodeInfo>, NodeId {
+public interface NodeInfo extends Comparable<NodeInfo> {
 
   static NodeInfo of(org.apache.kafka.common.Node node) {
     return of(node.id(), node.host(), node.port());
@@ -52,4 +52,7 @@ public interface NodeInfo extends Comparable<NodeInfo>, NodeId {
 
   /** @return The client (kafka data, jmx, etc.) port for this node */
   int port();
+
+  /** @return id of broker node. it must be unique. */
+  int id();
 }
diff --git a/app/src/main/java/org/astraea/partitioner/PartitionInfo.java b/app/src/main/java/org/astraea/partitioner/PartitionInfo.java
index 7e1b997d..197b77b5 100644
--- a/app/src/main/java/org/astraea/partitioner/PartitionInfo.java
+++ b/app/src/main/java/org/astraea/partitioner/PartitionInfo.java
@@ -3,20 +3,24 @@ package org.astraea.partitioner;
 public interface PartitionInfo {
 
   static PartitionInfo of(org.apache.kafka.common.PartitionInfo pf) {
+    return of(pf.topic(), pf.partition(), NodeInfo.of(pf.leader()));
+  }
+
+  static PartitionInfo of(String topic, int partition, NodeInfo leader) {
     return new PartitionInfo() {
       @Override
       public String topic() {
-        return pf.topic();
+        return topic;
       }
 
       @Override
       public int partition() {
-        return pf.partition();
+        return partition;
       }
 
       @Override
       public NodeInfo leader() {
-        return NodeInfo.of(pf.leader());
+        return leader;
       }
     };
   }
diff --git a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
index d5dbb3e1..d23be8d1 100644
--- a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
@@ -31,8 +31,8 @@ public class StrictCostDispatcher implements Dispatcher {
       BeanCollector.builder().interval(Duration.ofSeconds(4)).build();
   private final Collection<CostFunction> functions;
   private Optional<Integer> jmxPortDefault = Optional.empty();
-  private final Map<NodeId, Integer> jmxPorts = new TreeMap<>();
-  private final Map<NodeId, Receiver> receivers = new TreeMap<>();
+  private final Map<Integer, Integer> jmxPorts = new TreeMap<>();
+  final Map<Integer, Receiver> receivers = new TreeMap<>();
 
   public StrictCostDispatcher() {
     this(List.of(CostFunction.throughput()));
@@ -43,34 +43,22 @@ public class StrictCostDispatcher implements Dispatcher {
     this.functions = functions;
   }
 
-  private static Fetcher of(Collection<Fetcher> fs) {
-    return client ->
-        fs.stream().flatMap(f -> f.fetch(client).stream()).collect(Collectors.toUnmodifiableList());
-  }
-
   @Override
   public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
     var partitions = clusterInfo.availablePartitions(topic);
     // just return first partition if there is no available partitions
     if (partitions.isEmpty()) return 0;
 
+    // just return the only one available partition
+    if (partitions.size() == 1) return partitions.iterator().next().partition();
+
     // add new receivers for new brokers
     partitions.stream()
-        .filter(p -> !receivers.containsKey(NodeId.of(p.leader().id())))
+        .filter(p -> !receivers.containsKey(p.leader().id()))
         .forEach(
             p ->
                 receivers.put(
-                    p.leader(),
-                    beanCollector
-                        .register()
-                        .host(p.leader().host())
-                        .port(jmxPort(p.leader().id()))
-                        .fetcher(
-                            Fetcher.of(
-                                functions.stream()
-                                    .map(CostFunction::fetcher)
-                                    .collect(Collectors.toUnmodifiableList())))
-                        .build()));
+                    p.leader().id(), receiver(p.leader().host(), jmxPort(p.leader().id()))));
 
     // get latest beans for each node
     var beans =
@@ -83,15 +71,32 @@ public class StrictCostDispatcher implements Dispatcher {
             .map(f -> f.cost(beans, clusterInfo))
             .collect(Collectors.toUnmodifiableList());
 
-    // return the partition (node) having min score
+    return bestPartition(partitions, scores).map(e -> e.getKey().partition()).orElse(0);
+  }
+
+  // visible for testing
+  static Optional<Map.Entry<PartitionInfo, Double>> bestPartition(
+      List<PartitionInfo> partitions, List<Map<Integer, Double>> scores) {
     return partitions.stream()
         .map(
             p ->
                 Map.entry(
-                    p, scores.stream().mapToDouble(s -> s.getOrDefault(p.leader(), 0.0D)).sum()))
-        .min(Map.Entry.comparingByValue())
-        .map(e -> e.getKey().partition())
-        .orElse(0);
+                    p,
+                    scores.stream().mapToDouble(s -> s.getOrDefault(p.leader().id(), 0.0D)).sum()))
+        .min(Map.Entry.comparingByValue());
+  }
+
+  Receiver receiver(String host, int port) {
+    return beanCollector
+        .register()
+        .host(host)
+        .port(port)
+        .fetcher(
+            Fetcher.of(
+                functions.stream()
+                    .map(CostFunction::fetcher)
+                    .collect(Collectors.toUnmodifiableList())))
+        .build();
   }
 
   @Override
@@ -102,15 +107,12 @@ public class StrictCostDispatcher implements Dispatcher {
     config.entrySet().stream()
         .filter(e -> e.getKey().startsWith("broker."))
         .filter(e -> e.getKey().endsWith(JMX_PORT))
-        .forEach(
-            e ->
-                jmxPorts.put(
-                    NodeId.of(Integer.parseInt(e.getKey())), Integer.parseInt(e.getValue())));
+        .forEach(e -> jmxPorts.put(Integer.parseInt(e.getKey()), Integer.parseInt(e.getValue())));
   }
 
   // visible for testing
   int jmxPort(int id) {
-    if (jmxPorts.containsKey(NodeId.of(id))) return jmxPorts.get(NodeId.of(id));
+    if (jmxPorts.containsKey(id)) return jmxPorts.get(id);
     return jmxPortDefault.orElseThrow(
         () -> new NoSuchElementException("broker: " + id + " does not have jmx port"));
   }
@@ -118,5 +120,6 @@ public class StrictCostDispatcher implements Dispatcher {
   @Override
   public void close() {
     receivers.values().forEach(Utils::close);
+    receivers.clear();
   }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java b/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
index c22909da..81475d73 100644
--- a/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
@@ -5,7 +5,6 @@ import java.util.Map;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.partitioner.ClusterInfo;
-import org.astraea.partitioner.NodeId;
 
 public interface CostFunction {
 
@@ -20,7 +19,7 @@ public interface CostFunction {
    * @param clusterInfo cluster information
    * @return the score of each node. the score range is [0 - 1]
    */
-  Map<NodeId, Double> cost(Map<NodeId, List<HasBeanObject>> beans, ClusterInfo clusterInfo);
+  Map<Integer, Double> cost(Map<Integer, List<HasBeanObject>> beans, ClusterInfo clusterInfo);
 
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
   Fetcher fetcher();
diff --git a/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java b/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
index 46d20994..7464f92b 100644
--- a/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
+++ b/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
@@ -8,21 +8,23 @@ import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
 import org.astraea.metrics.kafka.KafkaMetrics;
 import org.astraea.partitioner.ClusterInfo;
-import org.astraea.partitioner.NodeId;
+import org.astraea.partitioner.NodeInfo;
 
 public class ThroughputCost implements CostFunction {
 
   @Override
-  public Map<NodeId, Double> cost(Map<NodeId, List<HasBeanObject>> beans, ClusterInfo clusterInfo) {
+  public Map<Integer, Double> cost(
+      Map<Integer, List<HasBeanObject>> beans, ClusterInfo clusterInfo) {
     var score = score(beans);
 
     var max = score.values().stream().mapToDouble(v -> v).max().orElse(1);
 
     return clusterInfo.nodes().stream()
+        .map(NodeInfo::id)
         .collect(Collectors.toMap(n -> n, n -> score.getOrDefault(n, 0.0D) / max));
   }
 
-  Map<NodeId, Double> score(Map<NodeId, List<HasBeanObject>> beans) {
+  Map<Integer, Double> score(Map<Integer, List<HasBeanObject>> beans) {
     // TODO: this implementation only consider the oneMinuteRate ...
     return beans.entrySet().stream()
         .collect(
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
index 34281d9c..861d309f 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -9,6 +9,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -20,6 +22,7 @@ import org.astraea.metrics.collector.Receiver;
 import org.astraea.metrics.java.HasJvmMemory;
 import org.astraea.metrics.kafka.KafkaMetrics;
 import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.NodeInfo;
 
 /**
  * this clas is responsible for obtaining jmx metrics from BeanCollector and calculating the
@@ -27,9 +30,8 @@ import org.astraea.partitioner.ClusterInfo;
  */
 public class NodeLoadClient {
 
-  private final List<Receiver> receiverList;
-  private final Map<String, Integer> currentJmxAddresses;
   private final Lock lock = new ReentrantLock();
+  private final Optional<Integer> jmxPortDefault;
   private static final ReceiverFactory RECEIVER_FACTORY = new ReceiverFactory();
   private long lastTime = -1;
   // visible for testing
@@ -38,14 +40,16 @@ public class NodeLoadClient {
   private int referenceBrokerID;
   private int totalInput;
   private int totalOutput;
+  private Map<Integer, Integer> jmxOfBrokerID;
+  private Map<String, Integer> jmxAddress;
 
   private static final String regex =
       "((25[0-5]|2[0-4]\\d|1\\d{2}|[1-9]?\\d)\\.){3}" + "(25[0-5]|2[0-4]\\d|1\\d{2}|[1-9]?\\d)$";
 
-  public NodeLoadClient(Map<String, Integer> jmxAddresses) {
-    this.receiverList = RECEIVER_FACTORY.receiversList(jmxAddresses);
-    receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
-    currentJmxAddresses = jmxAddresses;
+  public NodeLoadClient(Map<Integer, Integer> jmxAddresses, int jmxPortDefault) {
+    if (jmxPortDefault != -1) this.jmxPortDefault = Optional.of(jmxPortDefault);
+    else this.jmxPortDefault = Optional.empty();
+    jmxOfBrokerID = jmxAddresses;
   }
 
   /**
@@ -69,6 +73,29 @@ public class NodeLoadClient {
     return currentLoadNode;
   }
 
+  Map<String, Integer> jmxAddress(ClusterInfo cluster) {
+    var iterator = jmxOfBrokerID.keySet().iterator();
+    var currentJmxAddresses = new TreeMap<String, Integer>();
+    var idHost = cluster.nodes().stream().collect(Collectors.toMap(NodeInfo::id, NodeInfo::host));
+    while (iterator.hasNext()) {
+      var id = iterator.next();
+      if (idHost.containsKey(id)) {
+        currentJmxAddresses.put(ipAddress(idHost.get(id)), jmxOfBrokerID.get(id));
+        idHost.remove(id);
+      }
+    }
+
+    if (idHost.size() > 0) {
+      jmxPortDefault.ifPresentOrElse(
+          port -> idHost.forEach((key, value) -> currentJmxAddresses.put(ipAddress(value), port)),
+          () -> {
+            var notFindJMX = new ArrayList<>(idHost.keySet());
+            throw new RuntimeException(notFindJMX + "jmx servers not found.");
+          });
+    }
+    return currentJmxAddresses;
+  }
+
   /** @param cluster the cluster from the partitioner */
   private void nodesOverLoad(ClusterInfo cluster) {
     if (lastTime == -1) {
@@ -78,6 +105,9 @@ public class NodeLoadClient {
               .map(node -> new Broker(node.id(), node.host(), node.port()))
               .collect(Collectors.toList());
       referenceBrokerID = brokers.stream().findFirst().get().brokerID;
+      jmxAddress = jmxAddress(cluster);
+      var receiverList = RECEIVER_FACTORY.receiversList(jmxAddress);
+      receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
       var nodeIDReceiver = new HashMap<Integer, List<Receiver>>();
       receiverList.forEach(
           receiver -> {
@@ -137,7 +167,7 @@ public class NodeLoadClient {
               Long.parseLong(
                   broker
                       .metrics
-                      .get(KafkaMetrics.BrokerTopic.BytesInPerSec.name())
+                      .get(KafkaMetrics.BrokerTopic.BytesOutPerSec.name())
                       .current()
                       .get(0)
                       .beanObject()
@@ -236,7 +266,7 @@ public class NodeLoadClient {
   }
 
   public void close() {
-    RECEIVER_FACTORY.close(currentJmxAddresses);
+    RECEIVER_FACTORY.close(jmxAddress);
   }
 
   private List<Integer> brokerIDOfReceiver(String host) {
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
index 9036314c..b557cd74 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
@@ -1,7 +1,10 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import java.io.FileInputStream;
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 
 /** do poisson for node's load situation */
 public class PartitionerUtils {
@@ -45,20 +48,13 @@ public class PartitionerUtils {
     return (int) Math.round(avgLoadCount);
   }
 
-  /**
-   * A cheap way to deterministically convert a number to a positive value. When the input is
-   * positive, the original value is returned. When the input number is negative, the returned
-   * positive value is the original value bit AND against 0x7fffffff which is not its absolute
-   * value.
-   *
-   * <p>Note: changing this method in the future will possibly cause partition selection not to be
-   * compatible with the existing messages already placed on a partition since it is used in
-   * producer's {@link org.apache.kafka.clients.producer.internals.DefaultPartitioner}
-   *
-   * @param number a given number
-   * @return a positive number.
-   */
-  public static int toPositive(int number) {
-    return number & 0x7fffffff;
+  public static Properties partitionerConfig(Map<String, ?> configs) {
+    var properties = new Properties();
+    try {
+      properties.load(new FileInputStream((String) configs.get("partitioner.config")));
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    return properties;
   }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
index bdc8bc98..fb6f7ac6 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
@@ -69,9 +69,9 @@ public class ReceiverFactory {
     return receiversList;
   }
 
-  public void close(Map<String, Integer> jmxAddresses) {
+  public void close(Map<String, Integer> jmxAddress) {
     synchronized (lock) {
-      jmxAddresses.forEach(
+      jmxAddress.forEach(
           (host, port) -> {
             if (count.containsKey(nodeKey(host, port))) {
               if (count.get(nodeKey(host, port)) == 1) {
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
index b1632239..5d94d7a1 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
@@ -2,22 +2,23 @@ package org.astraea.partitioner.smoothPartitioner;
 
 import static org.astraea.Utils.overSecond;
 import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.allPoisson;
+import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.partitionerConfig;
 import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.weightPoisson;
 
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Random;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
 import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.Configuration;
 import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
 
 /**
@@ -26,14 +27,14 @@ import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
  * Finally, the result of poisson is used as the weight to perform smooth weighted RoundRobin.
  */
 public class SmoothWeightPartitioner implements Partitioner {
+  private static final String JMX_PORT = "jmx.port";
 
-  /**
-   * Record the current weight of each node according to Poisson calculation and the weight after
-   * partitioner calculation.
-   */
+  //  Record the current weight of each node according to Poisson calculation and the weight after
+  // partitioner calculation.
   private final ConcurrentMap<Integer, SmoothWeightServer> brokersWeight =
       new ConcurrentHashMap<>();
 
+  private final Map<Integer, Integer> jmxPorts = new TreeMap<>();
   private NodeLoadClient nodeLoadClient;
   private long lastTime = -1;
   private final Random rand = new Random();
@@ -49,7 +50,7 @@ public class SmoothWeightPartitioner implements Partitioner {
 
     SmoothWeightServer maxWeightServer = null;
 
-    Iterator<Integer> iterator = brokersWeight.keySet().iterator();
+    var iterator = brokersWeight.keySet().iterator();
     var currentWrightSum = weightSum;
     while (iterator.hasNext()) {
       var smoothWeightServer = brokersWeight.get(iterator.next());
@@ -80,19 +81,26 @@ public class SmoothWeightPartitioner implements Partitioner {
 
   @Override
   public void configure(Map<String, ?> configs) {
-    var jmxAddresses =
-        Objects.requireNonNull(
-            configs.get("jmx_servers").toString(), "You must configure jmx_servers correctly");
-    var list = Arrays.asList((jmxAddresses).split(","));
-    HashMap<String, Integer> mapAddress = new HashMap<>();
-    list.forEach(
-        str ->
-            mapAddress.put(
-                Arrays.asList(str.split(":")).get(0),
-                Integer.parseInt(Arrays.asList(str.split(":")).get(1))));
-    Objects.requireNonNull(mapAddress, "You must configure jmx_servers correctly.");
-
-    nodeLoadClient = new NodeLoadClient(mapAddress);
+    var properties = partitionerConfig(configs);
+    var config =
+        Configuration.of(
+            properties.entrySet().stream()
+                .collect(
+                    Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())));
+    var jmxPort = config.integer(JMX_PORT);
+
+    jmxPort = config.integer(JMX_PORT);
+
+    // seeks for custom jmx ports.
+    config.entrySet().stream()
+        .filter(e -> e.getKey().startsWith("broker."))
+        .filter(e -> e.getKey().endsWith(JMX_PORT))
+        .forEach(
+            e ->
+                jmxPorts.put(
+                    Integer.parseInt(e.getKey().split("\\.")[1]), Integer.parseInt(e.getValue())));
+
+    nodeLoadClient = new NodeLoadClient(jmxPorts, jmxPort.orElse(-1));
   }
 
   /** Change the weight of the node according to the current Poisson. */
diff --git a/app/src/main/java/org/astraea/performance/Distribution.java b/app/src/main/java/org/astraea/performance/Distribution.java
index 0942b253..83392220 100644
--- a/app/src/main/java/org/astraea/performance/Distribution.java
+++ b/app/src/main/java/org/astraea/performance/Distribution.java
@@ -5,14 +5,14 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
 import java.util.stream.IntStream;
-import org.astraea.argument.NonEmptyField;
+import org.astraea.argument.Field;
 
 /** Randomly generate a long number with respect to some distribution */
 public interface Distribution {
 
   long get();
 
-  class DistributionField implements NonEmptyField<Distribution> {
+  class DistributionField extends Field<Distribution> {
     @Override
     public Distribution convert(String rawArgument) {
       var args = rawArgument.split(":");
diff --git a/app/src/main/java/org/astraea/performance/ExeTime.java b/app/src/main/java/org/astraea/performance/ExeTime.java
index 07bc7f6e..945765aa 100644
--- a/app/src/main/java/org/astraea/performance/ExeTime.java
+++ b/app/src/main/java/org/astraea/performance/ExeTime.java
@@ -5,7 +5,6 @@ import java.time.Duration;
 import java.util.function.BiFunction;
 import java.util.regex.Pattern;
 import org.astraea.argument.DurationField;
-import org.astraea.argument.NonEmptyField;
 
 /**
  * Two kind of running modes. One runs for a duration of time. The other runs for a number of
@@ -38,11 +37,11 @@ interface ExeTime {
     };
   }
 
-  class Field implements NonEmptyField<ExeTime> {
+  class Field extends org.astraea.argument.Field<ExeTime> {
     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 {
+    protected void check(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/FileWriter.java b/app/src/main/java/org/astraea/performance/FileWriter.java
index f29e1756..b1ca1ed1 100644
--- a/app/src/main/java/org/astraea/performance/FileWriter.java
+++ b/app/src/main/java/org/astraea/performance/FileWriter.java
@@ -5,10 +5,11 @@ import java.io.IOException;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.stream.IntStream;
-import org.astraea.concurrent.ThreadPool;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
 import org.astraea.utils.DataUnit;
 
-public class FileWriter implements ThreadPool.Executor {
+public class FileWriter implements Executor {
   private final Manager manager;
   private final Tracker tracker;
   private BufferedWriter writer;
@@ -21,7 +22,7 @@ public class FileWriter implements ThreadPool.Executor {
   }
 
   @Override
-  public ThreadPool.Executor.State execute() throws InterruptedException {
+  public State execute() throws InterruptedException {
     if (logToCSV()) return State.DONE;
     Thread.sleep(1000);
     return State.RUNNING;
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 4ada2f00..58f1d924 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -3,15 +3,12 @@ package org.astraea.performance;
 import com.beust.jcommander.Parameter;
 import java.io.IOException;
 import java.time.Duration;
-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;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -23,9 +20,12 @@ import org.apache.kafka.common.record.CompressionType;
 import org.astraea.Utils;
 import org.astraea.argument.CompressionField;
 import org.astraea.argument.NonEmptyStringField;
-import org.astraea.argument.NonNegativeLongField;
+import org.astraea.argument.NonNegativeShortField;
 import org.astraea.argument.PositiveLongField;
 import org.astraea.argument.PositiveShortField;
+import org.astraea.argument.StringMapField;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.producer.Producer;
@@ -66,10 +66,9 @@ public class Performance {
     execute(org.astraea.argument.Argument.parse(new Argument(), args));
   }
 
-  public static Future<String> execute(final Argument param)
+  public static Result execute(final Argument param)
       throws InterruptedException, IOException, ExecutionException {
     List<Integer> partitions;
-    var future = new CompletableFuture<String>();
     try (var topicAdmin = TopicAdmin.of(param.props())) {
       topicAdmin
           .creator()
@@ -93,10 +92,8 @@ public class Performance {
 
     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();
     var groupId = "groupId-" + System.currentTimeMillis();
-    try (ThreadPool threadPool =
+    try (var threadPool =
         ThreadPool.builder()
             .executors(
                 IntStream.range(0, param.consumers)
@@ -129,17 +126,16 @@ public class Performance {
                                 manager))
                     .collect(Collectors.toUnmodifiableList()))
             .executor(tracker)
-            .executors(fileWriter)
+            .executor((param.createCSV) ? new FileWriter(manager, tracker) : () -> State.DONE)
             .build()) {
       threadPool.waitAll();
-      future.complete(param.topic);
-      return future;
+      return new Result(param.topic);
     }
   }
 
-  static ThreadPool.Executor consumerExecutor(
+  static Executor consumerExecutor(
       Consumer<byte[], byte[]> consumer, BiConsumer<Long, Long> observer, Manager manager) {
-    return new ThreadPool.Executor() {
+    return new Executor() {
       @Override
       public State execute() {
         try {
@@ -173,13 +169,13 @@ public class Performance {
     };
   }
 
-  static ThreadPool.Executor producerExecutor(
+  static Executor producerExecutor(
       Producer<byte[], byte[]> producer,
       Argument param,
       BiConsumer<Long, Long> observer,
       List<Integer> partitions,
       Manager manager) {
-    return new ThreadPool.Executor() {
+    return new Executor() {
       @Override
       public State execute() throws InterruptedException {
         // Wait for all consumers get assignment.
@@ -260,8 +256,8 @@ public class Performance {
     @Parameter(
         names = {"--consumers"},
         description = "Integer: number of consumers to consume records",
-        validateWith = NonNegativeLongField.class,
-        converter = NonNegativeLongField.class)
+        validateWith = NonNegativeShortField.class,
+        converter = NonNegativeShortField.class)
     int consumers = 1;
 
     @Parameter(
@@ -289,7 +285,7 @@ public class Performance {
     @Parameter(
         names = {"--jmx.servers"},
         description =
-            "String: server to get jmx metrics <jmx_server>@<broker_id>[,<jmx_server>@<broker_id>]*",
+            "String: server to get jmx metrics <broker_id>.<jmx_serverPort>[,<broker_id>.<jmx_serverPort>]*",
         validateWith = NonEmptyStringField.class)
     String jmxServers = "";
 
@@ -299,10 +295,18 @@ public class Performance {
         validateWith = NonEmptyStringField.class)
     String partitioner = DefaultPartitioner.class.getName();
 
+    @Parameter(
+        names = {"--configs"},
+        description = "Map: set the configuration passed to producer/partitioner",
+        converter = StringMapField.class,
+        validateWith = StringMapField.class)
+    Map<String, String> configs = Map.of();
+
     public Map<String, Object> producerProps() {
       var props = props();
       props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.name);
       if (!this.jmxServers.isEmpty()) props.put("jmx_servers", this.jmxServers);
+      props.putAll(configs);
       return props;
     }
 
@@ -332,4 +336,16 @@ public class Performance {
         validateWith = NonEmptyStringField.class)
     List<Integer> specifyBroker = List.of(-1);
   }
+
+  public static class Result {
+    private final String topicName;
+
+    private Result(String topicName) {
+      this.topicName = topicName;
+    }
+
+    public String topicName() {
+      return topicName;
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index d99c1742..9bbab9e2 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -4,11 +4,12 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import org.astraea.concurrent.ThreadPool;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
 import org.astraea.utils.DataUnit;
 
 /** Print out the given metrics. */
-public class Tracker implements ThreadPool.Executor {
+public class Tracker implements Executor {
   private final List<Metrics> producerData;
   private final List<Metrics> consumerData;
   private final Manager manager;
diff --git a/app/src/main/java/org/astraea/topic/TopicExplorer.java b/app/src/main/java/org/astraea/topic/TopicExplorer.java
index 79a56cff..c3b47c56 100644
--- a/app/src/main/java/org/astraea/topic/TopicExplorer.java
+++ b/app/src/main/java/org/astraea/topic/TopicExplorer.java
@@ -63,6 +63,21 @@ public class TopicExplorer {
 
   static Result execute(TopicAdmin admin, Set<String> topics) {
     var replicas = admin.replicas(topics);
+
+    var invalidTopics =
+        replicas.entrySet().stream()
+            .filter(x -> x.getValue().stream().noneMatch(Replica::leader))
+            .map(Map.Entry::getKey)
+            .sorted(
+                Comparator.comparing(TopicPartition::topic)
+                    .thenComparing(TopicPartition::partition))
+            .collect(Collectors.toUnmodifiableList());
+    if (!invalidTopics.isEmpty()) {
+      // The metadata request sent by Kafka admin succeeds only if the leaders of all partitions are
+      // alive. Hence, we throw exception here to make consistent behavior with Kafka.
+      throw new IllegalStateException("Some partitions have no leader: " + invalidTopics);
+    }
+
     var offsets = admin.offsets(topics);
     var consumerGroups = admin.consumerGroup(Set.of());
     var time = LocalDateTime.now();
diff --git a/app/src/main/java/org/astraea/utils/DataSize.java b/app/src/main/java/org/astraea/utils/DataSize.java
index ac3f15fc..98e9f774 100644
--- a/app/src/main/java/org/astraea/utils/DataSize.java
+++ b/app/src/main/java/org/astraea/utils/DataSize.java
@@ -10,7 +10,6 @@ import java.util.Comparator;
 import java.util.Objects;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import org.astraea.argument.NonEmptyField;
 
 /** Data size class */
 public class DataSize implements Comparable<DataSize> {
@@ -215,7 +214,7 @@ public class DataSize implements Comparable<DataSize> {
     return Objects.hash(bits);
   }
 
-  public static class Field implements NonEmptyField<DataSize> {
+  public static class Field extends org.astraea.argument.Field<DataSize> {
     // Parse number and DataUnit
     private static final Pattern DATA_SIZE_PATTERN =
         Pattern.compile("(?<measurement>[0-9]+)\\s?(?<dataUnit>[a-zA-Z]+)");
diff --git a/config/partitionerConfig.properties b/config/partitionerConfig.properties
new file mode 100644
index 00000000..e31beda9
--- /dev/null
+++ b/config/partitionerConfig.properties
@@ -0,0 +1,20 @@
+#The properties are used to configure SmoothWeightPartitioner
+#
+#If broker.id.jmx.port is configured, then the jmx port of these brokers
+# will be configured first. If there are brokers that are not configured
+# to jmx port, then jmx.port will be used for configuration. And if any
+# broker is not configured, and jmx.port is not configured at the same
+# time, then an error will be reported.
+#
+###########################Smooth Partitioner Basics############################
+
+#If no jmx port is specified for the broker,jmx.port will be used.
+jmx.port=Default;
+
+# list of brokers used for jmx port about the rest of the cluster
+# format: broker.{brokerID}.jmx.port=jmx.port
+#broker.0.jmx.port=9999
+#broker.1.jmx.port=8888
+#......
+bootstrap.servers=localhost:9092
+broker.id.jmx.port=Default;
diff --git a/docker/start_node_exporter.sh b/docker/start_node_exporter.sh
index 30b30a34..703bf5b0 100755
--- a/docker/start_node_exporter.sh
+++ b/docker/start_node_exporter.sh
@@ -56,10 +56,10 @@ function run_node_exporter() {
 container_id="$(run_node_exporter)"
 info Container ID of node_exporter: "$container_id"
 
-if [[ "$(curl --connect-timeout 1 --retry 3 -s "http://0.0.0.0:$PORT")" == "" ]] || [[ "$(docker_container_running "$container_id")" == "false" ]]; then
-  error Cannot access node_exporter metric server: "http://0.0.0.0:$PORT". Something go wrong!
+if curl --connect-timeout 1 --retry 3 --retry-connrefused -s "http://$address:$PORT/metrics" > /dev/null; then
+  info node_exporter running at "http://$address:$PORT"
+else
+  error Cannot access node_exporter metric server: "http://$address:$PORT/metrics". Something go wrong!
   error Execute \"docker logs "$container_id"\" to see what\'s going on.
   exit 3
 fi
-
-info node_exporter running at "http://$address:$PORT"
\ No newline at end of file
diff --git a/docker/start_prometheus.sh b/docker/start_prometheus.sh
index c1e61511..de0bacd4 100755
--- a/docker/start_prometheus.sh
+++ b/docker/start_prometheus.sh
@@ -7,9 +7,33 @@ function showHelp() {
   echo "COMMAND: "
   echo "    start <kafka-broker-jmx-addresses> <node-exporter-addresses>    create/start the prometheus docker instance"
   echo "    refresh <kafka-broker-jmx-addresses> <node-exporter-addresses>  refresh and apply the prometheus config"
+  echo "    refresh <config-file>                                           refresh and apply the prometheus config"
   echo "    help                                                            show this dialog"
 }
 
+function gradle_project_directory {
+    local script_folder="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
+    local target_folder="$script_folder"
+
+    # attempt to find the nearest parent directory that contains the gradlew file
+    # we recognize such directory as a Gradle repo that contains the build directory
+    while [ "$target_folder" != "/" ]; do
+        target_folder="$(dirname $target_folder)"
+
+        if [ -f "$target_folder/gradlew" ]; then
+            break
+        fi
+    done
+
+    if [ "$target_folder" == "/" ]; then
+        echo "Cannot find the gradle directory from $script_folder."
+        echo "Is the directory exists?"
+        exit 5
+    fi
+
+    echo "$target_folder"
+}
+
 # ===============================[checks]===============================
 
 if [[ "$(which docker)" == "" ]]; then
@@ -46,11 +70,16 @@ function wrap_address() {
 prometheus_port="${PROMETHEUS_PORT:-9090}"
 image_name=prom/prometheus:v2.32.1
 container_name="prometheus-${prometheus_port}"
-file=/tmp/prometheus-${prometheus_port}.yml
-scrape_interval="5s"
+file_prefix="$(gradle_project_directory)/build/docker/prometheus"
+file="${file_prefix}/prometheus-${prometheus_port}.yml"
+temp_file="${file_prefix}/prometheus-${prometheus_port}-editing.yml"
+scrape_interval="${scrape_interval:-200ms}"
+scrape_timeout="${scrape_timeout:-200ms}"
 volume_name_1="prometheus-${prometheus_port}-etc"
 volume_name_2="prometheus-${prometheus_port}-prometheus"
 
+mkdir -p "$file_prefix"
+
 function write_config() {
   kafka_jmx_addresses="$(wrap_address "$1")"
   node_exporter_addresses="$(wrap_address "$2")"
@@ -65,11 +94,13 @@ scrape_configs:
 
   - job_name: 'kafka'
     scrape_interval: $scrape_interval
+    scrape_timeout: $scrape_timeout
     static_configs:
       - targets: [$kafka_jmx_addresses]
 
   - job_name: 'node'
     scrape_interval: $scrape_interval
+    scrape_timeout: $scrape_timeout
     static_configs:
       - targets: [$node_exporter_addresses]
 EOT
@@ -91,8 +122,8 @@ function info() {
 }
 
 function refresh_config_from_file() {
-  info "Refresh prometheus configuration from config file $file"
-  docker cp "$file" "$container_name:/etc/prometheus/prometheus.yml"
+  info "Refresh prometheus configuration from config file $1"
+  docker cp "$1" "$container_name:/etc/prometheus/prometheus.yml"
   docker kill --signal="SIGHUP" "$container_name"
 }
 
@@ -134,19 +165,18 @@ function main() {
   info "================================================="
 }
 
-
 # shellcheck disable=SC2199
 if [[ "$1" == "start" ]]; then
   # shellcheck disable=SC2068
   main "$2" "$3"
-elif [[ "$1" == "refresh" ]] && [[ "$2" != "" ]]; then
+elif [[ "$1" == "refresh" ]] && [[ "$#" -eq 3 ]]; then
   # shellcheck disable=SC2068
   refresh_config "$2" "$3"
-elif [[ "$1" == "refresh" ]]; then
-  refresh_config_from_file
+elif [[ "$1" == "refresh" ]] && [[ "$#" -eq 2 ]]; then
+  refresh_config_from_file "$2"
 elif [[ "$1" == "help" ]]; then
   showHelp
 else
   info Unknown argument: "$@"
   showHelp
-fi
\ No newline at end of file
+fi


Test patch:
diff --git a/app/src/test/java/org/astraea/argument/FieldTest.java b/app/src/test/java/org/astraea/argument/FieldTest.java
new file mode 100644
index 00000000..c4cd2868
--- /dev/null
+++ b/app/src/test/java/org/astraea/argument/FieldTest.java
@@ -0,0 +1,22 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FieldTest {
+
+  @Test
+  void testNullAndEmptyValue() {
+    var f =
+        new Field<String>() {
+          @Override
+          public String convert(String value) {
+            return null;
+          }
+        };
+    Assertions.assertThrows(ParameterException.class, () -> f.validate("a", ""));
+    Assertions.assertThrows(ParameterException.class, () -> f.validate("a", null));
+    Assertions.assertThrows(ParameterException.class, () -> f.validate("a", "   "));
+  }
+}
diff --git a/app/src/test/java/org/astraea/argument/StringMapFieldTest.java b/app/src/test/java/org/astraea/argument/StringMapFieldTest.java
new file mode 100644
index 00000000..e92e5c15
--- /dev/null
+++ b/app/src/test/java/org/astraea/argument/StringMapFieldTest.java
@@ -0,0 +1,24 @@
+package org.astraea.argument;
+
+import com.beust.jcommander.ParameterException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class StringMapFieldTest {
+
+  @Test
+  void testStringInMap() {
+    var field = new StringMapField();
+    var result = field.convert("k0=v0,k1=v1");
+    Assertions.assertEquals(2, result.size());
+    Assertions.assertEquals("v0", result.get("k0"));
+    Assertions.assertEquals("v1", result.get("k1"));
+  }
+
+  @Test
+  void testNonMap() {
+    var field = new StringMapField();
+    Assertions.assertThrows(ParameterException.class, () -> field.validate("a", "bb"));
+    Assertions.assertThrows(ParameterException.class, () -> field.validate("a", "bb="));
+  }
+}
diff --git a/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java b/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
index c973347c..ab3a15ad 100644
--- a/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
+++ b/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
@@ -8,7 +8,7 @@ import org.junit.jupiter.api.Timeout;
 
 public class ThreadPoolTest {
 
-  private static class CountExecutor implements ThreadPool.Executor {
+  private static class CountExecutor implements Executor {
 
     private final AtomicInteger executeCount = new AtomicInteger();
     private final AtomicInteger closeCount = new AtomicInteger();
@@ -44,7 +44,7 @@ public class ThreadPoolTest {
 
   @Test
   void testWaitAll() {
-    try (var pool = ThreadPool.builder().executor(() -> ThreadPool.Executor.State.DONE).build()) {
+    try (var pool = ThreadPool.builder().executor(() -> State.DONE).build()) {
       pool.waitAll();
     }
   }
@@ -57,7 +57,7 @@ public class ThreadPoolTest {
             .executor(
                 () -> {
                   TimeUnit.SECONDS.sleep(1000);
-                  return ThreadPool.Executor.State.DONE;
+                  return State.DONE;
                 })
             .build();
     pool.close();
diff --git a/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java b/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
index ce50a8d6..a573747c 100644
--- a/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
+++ b/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
@@ -4,6 +4,7 @@ import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.astraea.Utils;
+import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Test;
@@ -19,12 +20,12 @@ public class RebalanceListenerTest extends RequireBrokerCluster {
             .topics(Set.of(topicName))
             .consumerRebalanceListener(ignore -> getAssignment.incrementAndGet())
             .build()) {
-      try (ThreadPool threadPool =
+      try (var threadPool =
           ThreadPool.builder()
               .executor(
                   () -> {
                     consumer.poll(Duration.ofSeconds(10));
-                    return ThreadPool.Executor.State.DONE;
+                    return State.DONE;
                   })
               .build()) {
         Utils.waitFor(() -> getAssignment.get() == 1, Duration.ofSeconds(10));
diff --git a/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
index 801268e9..522383a3 100644
--- a/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
+++ b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
@@ -9,6 +9,8 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.jmx.MBeanClient;
@@ -128,7 +130,10 @@ public class BeanCollectorTest {
 
     try (var pool =
         ThreadPool.builder()
-            .runnables(IntStream.range(0, 3).mapToObj(i -> runnable).collect(Collectors.toList()))
+            .executors(
+                IntStream.range(0, 3)
+                    .mapToObj(i -> executor(runnable))
+                    .collect(Collectors.toList()))
             .build()) {
       sleep(1);
     }
@@ -164,9 +169,9 @@ public class BeanCollectorTest {
 
     try (var pool =
         ThreadPool.builder()
-            .runnables(
+            .executors(
                 receivers.stream()
-                    .map(receiver -> ((Runnable) receiver::current))
+                    .map(receiver -> executor(receiver::current))
                     .collect(Collectors.toList()))
             .build()) {
       sleep(3);
@@ -241,4 +246,11 @@ public class BeanCollectorTest {
     receivers.forEach(e -> Assertions.assertEquals(1, e.getValue().current().size()));
     receivers.forEach(e -> Assertions.assertEquals(1, e.getKey().get()));
   }
+
+  private static Executor executor(Runnable runnable) {
+    return () -> {
+      runnable.run();
+      return State.RUNNING;
+    };
+  }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
index fd7a03eb..f194629a 100644
--- a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
@@ -1,16 +1,28 @@
 package org.astraea.partitioner;
 
+import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.collector.Fetcher;
+import org.astraea.metrics.collector.Receiver;
+import org.astraea.partitioner.cost.CostFunction;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 public class StrictCostDispatcherTest {
 
   @Test
   void testClose() {
     var dispatcher = new StrictCostDispatcher();
+    var receiver = Mockito.mock(Receiver.class);
+    dispatcher.receivers.put(10, receiver);
     dispatcher.close();
+    Mockito.verify(receiver, Mockito.times(1)).close();
+    Assertions.assertEquals(0, dispatcher.receivers.size());
   }
 
   @Test
@@ -22,4 +34,74 @@ public class StrictCostDispatcherTest {
       Assertions.assertEquals(12345, dispatcher.jmxPort(0));
     }
   }
+
+  @Test
+  void testBestPartition() {
+    var partitions =
+        List.of(
+            PartitionInfo.of("t", 0, NodeInfo.of(10, "h0", 1000)),
+            PartitionInfo.of("t", 1, NodeInfo.of(11, "h1", 1000)));
+
+    var score = List.of(Map.of(10, 0.8D), Map.of(11, 0.7D));
+
+    var result = StrictCostDispatcher.bestPartition(partitions, score).get();
+
+    Assertions.assertEquals(0.7D, result.getValue());
+    Assertions.assertEquals(1, result.getKey().partition());
+  }
+
+  @Test
+  void testPartition() {
+    // n0 is busy
+    var n0 = NodeInfo.of(10, "host0", 12345);
+    var p0 = PartitionInfo.of("aa", 1, n0);
+    // n1 is free
+    var n1 = NodeInfo.of(11, "host1", 12345);
+    var p1 = PartitionInfo.of("aa", 2, n1);
+
+    var receiver = Mockito.mock(Receiver.class);
+    Mockito.when(receiver.current()).thenReturn(List.of());
+
+    var costFunction =
+        new CostFunction() {
+          @Override
+          public Map<Integer, Double> cost(
+              Map<Integer, List<HasBeanObject>> beans, ClusterInfo clusterInfo) {
+            return beans.keySet().stream()
+                .collect(
+                    Collectors.toMap(Function.identity(), id -> id.equals(n0.id()) ? 0.9D : 0.5D));
+          }
+
+          @Override
+          public Fetcher fetcher() {
+            return client -> List.of();
+          }
+        };
+
+    var dispatcher =
+        new StrictCostDispatcher(List.of(costFunction)) {
+          @Override
+          Receiver receiver(String host, int port) {
+            return receiver;
+          }
+
+          @Override
+          int jmxPort(int id) {
+            return 0;
+          }
+        };
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+
+    // there is no available partition
+    Mockito.when(clusterInfo.availablePartitions("aa")).thenReturn(List.of());
+    Assertions.assertEquals(0, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
+
+    // there is only one available partition
+    Mockito.when(clusterInfo.availablePartitions("aa")).thenReturn(List.of(p0));
+    Assertions.assertEquals(1, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
+
+    // there is no beans, so it just returns the partition.
+    Mockito.when(clusterInfo.availablePartitions("aa")).thenReturn(List.of(p0, p1));
+    Assertions.assertEquals(2, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
+  }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java b/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
index 53ab3b20..48cb0d9d 100644
--- a/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
@@ -19,9 +19,9 @@ public class ThroughputCostTest {
     var bean = Mockito.mock(BrokerTopicMetricsResult.class);
     Mockito.when(bean.oneMinuteRate()).thenReturn(100D);
 
-    var score = throughputCost.score(Map.of(node, List.of(bean)));
+    var score = throughputCost.score(Map.of(10, List.of(bean)));
     Assertions.assertEquals(1, score.size());
-    Assertions.assertEquals(100D, score.get(node));
+    Assertions.assertEquals(100D, score.get(10));
   }
 
   @Test
@@ -35,8 +35,8 @@ public class ThroughputCostTest {
     var cluster = Mockito.mock(ClusterInfo.class);
     Mockito.when(cluster.nodes()).thenReturn(List.of(node));
 
-    var cost = throughputCost.cost(Map.of(node, List.of(bean)), cluster);
+    var cost = throughputCost.cost(Map.of(10, List.of(bean)), cluster);
     Assertions.assertEquals(1, cost.size());
-    Assertions.assertEquals(1, cost.get(node));
+    Assertions.assertEquals(1, cost.get(10));
   }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
index 3c754173..4b85e90a 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
@@ -2,7 +2,6 @@ package org.astraea.partitioner.nodeLoadMetric;
 
 import static org.mockito.Mockito.when;
 
-import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -24,10 +23,10 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
   private NodeLoadClient nodeLoadClient;
 
   @BeforeAll
-  void setUp() throws IOException {
-    var map = new HashMap<String, Integer>();
-    map.put(jmxServiceURL().getHost(), jmxServiceURL().getPort());
-    nodeLoadClient = new NodeLoadClient(map);
+  void setUp() {
+    var map = new HashMap<Integer, Integer>();
+    map.put(0, jmxServiceURL().getPort());
+    nodeLoadClient = new NodeLoadClient(map, jmxServiceURL().getPort());
   }
 
   @AfterAll
@@ -89,7 +88,6 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
           nodes.add(NodeInfo.of(count.get(), hostPort[0], Integer.parseInt(hostPort[1])));
           count.getAndIncrement();
         });
-
     var cluster = Mockito.mock(ClusterInfo.class);
     when(cluster.nodes()).thenReturn(nodes);
     var load = nodeLoadClient.loadSituation(cluster);
diff --git a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
index 6fc9cf4e..c655bfa4 100644
--- a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
@@ -6,6 +6,9 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
 import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.HashMap;
@@ -20,6 +23,8 @@ import java.util.stream.IntStream;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.astraea.Utils;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
@@ -47,12 +52,24 @@ public class PartitionerTest extends RequireBrokerCluster {
     props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
     props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
     props.put("producerID", 1);
-    props.put("jmx_servers", jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort());
+    var file = new File(PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write("jmx.port=" + jmxServiceURL().getPort());
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    props.put(
+        "partitioner.config",
+        PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
     return props;
   }
 
   @Test
   void testSetBrokerHashMap() {
+    initProConfig();
     var nodeLoadClient = Mockito.mock(NodeLoadClient.class);
     when(nodeLoadClient.thoughPutComparison(anyInt())).thenReturn(1.0);
     var poissonMap = new HashMap<Integer, Double>();
@@ -135,7 +152,7 @@ public class PartitionerTest extends RequireBrokerCluster {
     var key = "tainan";
     var timestamp = System.currentTimeMillis() + 10;
     var header = Header.of("a", "b".getBytes());
-    try (ThreadPool threadPool =
+    try (var threadPool =
         ThreadPool.builder()
             .executors(
                 IntStream.range(0, 10)
@@ -216,9 +233,60 @@ public class PartitionerTest extends RequireBrokerCluster {
     Assertions.assertEquals(smoothWeightServer.originalWeight(), -9);
   }
 
-  private ThreadPool.Executor producerExecutor(
+  @Test
+  void testJmxConfig() {
+    var props = initProConfig();
+    var file = new File(PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write(
+          "broker.0.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.1.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.2.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n");
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    var topicName = "addressN";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+
+    try (var producer =
+        Producer.builder()
+            .keySerializer(Serializer.STRING)
+            .configs(
+                props.entrySet().stream()
+                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+            .build()) {
+      var metadata =
+          producer
+              .sender()
+              .topic(topicName)
+              .key(key)
+              .timestamp(timestamp)
+              .headers(List.of(header))
+              .run()
+              .toCompletableFuture()
+              .get();
+      assertEquals(topicName, metadata.topic());
+      assertEquals(timestamp, metadata.timestamp());
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+
+  private Executor producerExecutor(
       Producer<String, byte[]> producer, String topic, String key, Header header, long timeStamp) {
-    return new ThreadPool.Executor() {
+    return new Executor() {
       int i = 0;
 
       @Override
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 852ab5ae..a753be5e 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -2,12 +2,14 @@ package org.astraea.performance;
 
 import static org.astraea.performance.Performance.partition;
 
+import com.beust.jcommander.ParameterException;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.TopicPartition;
 import org.astraea.Utils;
+import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.producer.Producer;
@@ -34,7 +36,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     param.specifyBroker = List.of(0);
     param.consumers = 0;
     param.partitions = 10;
-    try (ThreadPool.Executor executor =
+    try (var executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
             param,
@@ -77,7 +79,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     param.specifyBroker = List.of(0, 1);
     param.consumers = 0;
     param.partitions = 10;
-    try (ThreadPool.Executor executor =
+    try (Executor executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
             param,
@@ -112,7 +114,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     param.topic = "testProducerExecutor-" + System.currentTimeMillis();
     param.fixedSize = true;
     param.consumers = 0;
-    try (ThreadPool.Executor executor =
+    try (Executor executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
             param,
@@ -132,7 +134,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
     var param = new Performance.Argument();
     param.fixedSize = true;
-    try (ThreadPool.Executor executor =
+    try (Executor executor =
         Performance.consumerExecutor(
             Consumer.builder().topics(Set.of(topicName)).brokers(bootstrapServers()).build(),
             metrics,
@@ -151,4 +153,100 @@ public class PerformanceTest extends RequireBrokerCluster {
       Assertions.assertNotEquals(1024, metrics.bytes());
     }
   }
+
+  @Test
+  void testArgument() {
+    String[] arguments1 = {
+      "--bootstrap.servers",
+      "localhost:9092",
+      "--topic",
+      "not-empty",
+      "--partitions",
+      "10",
+      "--replicas",
+      "3",
+      "--producers",
+      "1",
+      "--consumers",
+      "1",
+      "--run.until",
+      "1000records",
+      "--record.size",
+      "10KiB",
+      "--jmx.servers",
+      "localhost:9000@1",
+      "--partitioner",
+      "org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner",
+      "--compression",
+      "lz4",
+      "--key.distribution",
+      "zipfian",
+      "--specify.broker",
+      "1",
+      "--configs",
+      "key=value"
+    };
+
+    var arg = org.astraea.argument.Argument.parse(new Performance.Argument(), arguments1);
+    Assertions.assertEquals("value", arg.producerProps().get("key").toString());
+
+    String[] arguments2 = {"--bootstrap.servers", "localhost:9092", "--topic", ""};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments2));
+
+    String[] arguments3 = {"--bootstrap.servers", "localhost:9092", "--replicas", "0"};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments3));
+
+    String[] arguments4 = {"--bootstrap.servers", "localhost:9092", "--partitions", "0"};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments4));
+
+    String[] arguments5 = {"--bootstrap.servers", "localhost:9092", "--producers", "0"};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments5));
+
+    String[] arguments6 = {"--bootstrap.servers", "localhost:9092", "--consumers", "0"};
+    Assertions.assertDoesNotThrow(
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments6));
+
+    String[] arguments7 = {"--bootstrap.servers", "localhost:9092", "--run.until", "1"};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments7));
+
+    String[] arguments8 = {"--bootstrap.servers", "localhost:9092", "--record.size", "1"};
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments8));
+
+    String[] arguments9 = {"--bootstrap.servers", "localhost:9092", "--jmx.servers", ""};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments9));
+
+    String[] arguments10 = {"--bootstrap.servers", "localhost:9092", "--partitioner", ""};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments10));
+
+    String[] arguments11 = {"--bootstrap.servers", "localhost:9092", "--compression", ""};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments11));
+
+    String[] arguments12 = {"--bootstrap.servers", "localhost:9092", "--key.distribution", ""};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments12));
+
+    String[] arguments13 = {"--bootstrap.servers", "localhost:9092", "--specify.broker", ""};
+    Assertions.assertThrows(
+        ParameterException.class,
+        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments13));
+  }
 }
diff --git a/app/src/test/java/org/astraea/performance/TrackerTest.java b/app/src/test/java/org/astraea/performance/TrackerTest.java
index e725e321..4e3e75dd 100644
--- a/app/src/test/java/org/astraea/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/performance/TrackerTest.java
@@ -1,7 +1,7 @@
 package org.astraea.performance;
 
 import java.util.List;
-import org.astraea.concurrent.ThreadPool;
+import org.astraea.concurrent.State;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -16,21 +16,21 @@ public class TrackerTest {
 
     var manager = new Manager(argument, producerData, consumerData);
     try (Tracker tracker = new Tracker(producerData, consumerData, manager)) {
-      Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
+      Assertions.assertEquals(State.RUNNING, tracker.execute());
       producerData.get(0).accept(1L, 1L);
       consumerData.get(0).accept(1L, 1L);
       manager.producerClosed();
-      Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
+      Assertions.assertEquals(State.DONE, tracker.execute());
     }
 
     // Zero consumer
     producerData = List.of(new Metrics());
     manager = new Manager(argument, producerData, empty);
     try (Tracker tracker = new Tracker(producerData, empty, manager)) {
-      Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
+      Assertions.assertEquals(State.RUNNING, tracker.execute());
       producerData.get(0).accept(1L, 1L);
       manager.producerClosed();
-      Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
+      Assertions.assertEquals(State.DONE, tracker.execute());
     }
 
     // Stop by duration time out
@@ -40,7 +40,7 @@ public class TrackerTest {
     manager = new Manager(argument, producerData, consumerData);
     try (Tracker tracker = new Tracker(producerData, consumerData, manager)) {
       tracker.start = System.currentTimeMillis();
-      Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
+      Assertions.assertEquals(State.RUNNING, tracker.execute());
 
       // Mock record producing
       producerData.get(0).accept(1L, 1L);
@@ -48,7 +48,7 @@ public class TrackerTest {
       Thread.sleep(2000);
       manager.producerClosed();
 
-      Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
+      Assertions.assertEquals(State.DONE, tracker.execute());
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/topic/TopicExplorerTest.java b/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
index d64e1f95..6edefa65 100644
--- a/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
@@ -1,6 +1,7 @@
 package org.astraea.topic;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.ByteArrayOutputStream;
@@ -23,6 +24,7 @@ import org.astraea.consumer.Consumer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 public class TopicExplorerTest extends RequireBrokerCluster {
 
@@ -201,4 +203,30 @@ public class TopicExplorerTest extends RequireBrokerCluster {
     assertTrue(output.matches("(?ms).+\\[.*future.*].+"), "descriptor printed");
     assertTrue(output.matches("(?ms).+/tmp/path0.+"), "path printed");
   }
+
+  @Test
+  void somePartitionsHaveNoLeader() {
+    String topicName0 = "poor_topic0";
+    String topicName1 = "poor_topic1";
+    var payload0 =
+        Map.of(
+            new TopicPartition(topicName0, 0),
+            List.of(new Replica(1000, 0, 0, false, false, false, "?")));
+    var payload1 =
+        Map.of(
+            new TopicPartition(topicName1, 0),
+            List.of(
+                new Replica(1000, 0, 0, false, false, false, "?"),
+                new Replica(1001, 0, 0, false, false, false, "?"),
+                new Replica(1002, 0, 0, false, false, false, "?")));
+    TopicAdmin mock = Mockito.mock(TopicAdmin.class);
+
+    Mockito.when(mock.replicas(Set.of(topicName0))).thenReturn(payload0);
+    Mockito.when(mock.replicas(Set.of(topicName1))).thenReturn(payload1);
+
+    assertThrows(
+        IllegalStateException.class, () -> TopicExplorer.execute(mock, Set.of(topicName0)));
+    assertThrows(
+        IllegalStateException.class, () -> TopicExplorer.execute(mock, Set.of(topicName1)));
+  }
 }