Repository: skiptests/astraea (java)

Issue:
合併 BalancerHandler#parseCostFunctionWeight and StrictCostDispatcher#parseCostFunctionWeight
我們需要消除這兩段一樣的程式碼，簡單的想法如下：
1. 在`HasBrokerCost`, `HasClusterCost`中放置類似的邏輯，也就是吃`Map<String, String>`然後回傳`Map<Cost, Double>`
2. 上述方法預設將錯誤吃掉，但也留一個空間可以拋出錯誤

Pull request description:
[COMMON] Json converter should support DataSize and Duration
`Duration`和`DataSize`很常用在request/response，因此我們直接支援該型別的轉換來減少程式碼

fix #1121

Patch:
diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
index a2ba14df..aed6d7e7 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -19,13 +19,13 @@ package org.astraea.app.web;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.net.InetSocketAddress;
 import java.time.Duration;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -38,8 +38,8 @@ import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-import org.astraea.app.argument.DataSizeField;
 import org.astraea.common.Configuration;
+import org.astraea.common.DataSize;
 import org.astraea.common.EnumInfo;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
@@ -108,7 +108,7 @@ class BalancerHandler implements Handler {
     var planId = channel.target().get();
     if (!planCalculation.containsKey(planId))
       return CompletableFuture.completedFuture(Response.NOT_FOUND);
-    var timeout = requestHistory.get(planId).executionTime.toMillis();
+    var timeout = requestHistory.get(planId).executionTime;
     var balancer = requestHistory.get(planId).balancerClasspath;
     var functions = requestHistory.get(planId).algorithmConfig.clusterCostFunction().toString();
 
@@ -294,108 +294,85 @@ class BalancerHandler implements Handler {
   // visible for test
   static PostRequestWrapper parsePostRequestWrapper(
       BalancerPostRequest balancerPostRequest, ClusterInfo currentClusterInfo) {
-
-    var balancerClasspath = balancerPostRequest.balancer;
-    var balancerConfig = Configuration.of(balancerPostRequest.balancerConfig);
-    var clusterCostFunction = getClusterCost(balancerPostRequest);
-    var timeout = Utils.toDuration(balancerPostRequest.timeout);
-
     var topics =
-        balancerPostRequest
-            .topics
-            .map(
-                s ->
-                    Arrays.stream(s.split(","))
-                        .filter(x -> !x.isEmpty())
-                        .collect(Collectors.toSet()))
-            .orElseGet(currentClusterInfo::topics);
-
-    if (balancerPostRequest.topics.isPresent() && topics.isEmpty())
+        balancerPostRequest.topics.isEmpty()
+            ? currentClusterInfo.topics()
+            : balancerPostRequest.topics;
+
+    if (topics.isEmpty())
       throw new IllegalArgumentException(
           "Illegal topic filter, empty topic specified so nothing can be rebalance. ");
-    if (timeout.isZero() || timeout.isNegative())
+    if (balancerPostRequest.timeout.isZero() || balancerPostRequest.timeout.isNegative())
       throw new IllegalArgumentException(
-          "Illegal timeout, value should be positive integer: " + timeout.getSeconds());
+          "Illegal timeout, value should be positive integer: "
+              + balancerPostRequest.timeout.getSeconds());
 
     return new PostRequestWrapper(
-        balancerClasspath,
-        timeout,
+        balancerPostRequest.balancer,
+        balancerPostRequest.timeout,
         AlgorithmConfig.builder()
-            .clusterCost(clusterCostFunction)
+            .clusterCost(balancerPostRequest.clusterCost())
             .moveCost(DEFAULT_MOVE_COST_FUNCTIONS)
             .movementConstraint(movementConstraint(balancerPostRequest))
             .topicFilter(topics::contains)
-            .config(balancerConfig)
+            .config(Configuration.of(balancerPostRequest.balancerConfig))
             .build(),
         currentClusterInfo);
   }
 
   // TODO: There needs to be a way for"GU" and Web to share this function.
   static Predicate<MoveCost> movementConstraint(BalancerPostRequest request) {
-    var converter = new DataSizeField();
-    var replicaSizeLimit = request.maxMigratedSize.map(x -> converter.convert(x).bytes());
-    var leaderNumLimit = request.maxMigratedLeader.map(Integer::parseInt);
-    return cost ->
-        replicaSizeLimit
-                .filter(
-                    limit ->
-                        limit
-                            <= cost.movedReplicaSize().values().stream()
-                                .mapToLong(s -> Math.abs(s.bytes()))
-                                .sum())
-                .isEmpty()
-            && leaderNumLimit
-                .filter(
-                    limit ->
-                        limit
-                            <= cost.changedReplicaLeaderCount().values().stream()
-                                .mapToLong(s -> s)
-                                .sum())
-                .isEmpty();
-  }
-
-  static HasClusterCost getClusterCost(BalancerPostRequest request) {
-    var costWeights = request.costWeights;
-    if (costWeights.isEmpty()) throw new IllegalArgumentException("costWeights is not specified");
-    var costWeightMap =
-        costWeights.stream()
-            .flatMap(cw -> cw.cost.map(c -> Map.entry(c, cw.weight.orElse(1D))).stream())
-            .collect(
-                Collectors.toMap(
-                    e -> {
-                      try {
-                        var clz = Class.forName(e.getKey());
-                        if (!HasClusterCost.class.isAssignableFrom(clz))
-                          throw new IllegalArgumentException(
-                              "the class: " + e.getKey() + " is not sub class of HasClusterCost");
-                        return (HasClusterCost) Utils.construct(clz, Configuration.EMPTY);
-                      } catch (ClassNotFoundException error) {
-                        throw new IllegalArgumentException(error.getMessage());
-                      }
-                    },
-                    Map.Entry::getValue));
-    return HasClusterCost.of(costWeightMap);
+    return cost -> {
+      if (request.maxMigratedSize.bytes()
+          < cost.movedReplicaSize().values().stream().mapToLong(DataSize::bytes).sum())
+        return false;
+      if (request.maxMigratedLeader
+          < cost.changedReplicaLeaderCount().values().stream().mapToLong(s -> s).sum())
+        return false;
+      return true;
+    };
   }
 
   static class BalancerPostRequest implements Request {
 
-    private String balancer = GreedyBalancer.class.getName();
+    String balancer = GreedyBalancer.class.getName();
 
     Map<String, String> balancerConfig = Map.of();
 
-    String timeout = "3s";
-    Optional<String> topics = Optional.empty();
+    Duration timeout = Duration.ofSeconds(3);
+    Set<String> topics = Set.of();
 
-    private Optional<String> maxMigratedSize = Optional.empty();
+    DataSize maxMigratedSize = DataSize.Byte.of(Long.MAX_VALUE);
 
-    private Optional<String> maxMigratedLeader = Optional.empty();
+    long maxMigratedLeader = Long.MAX_VALUE;
 
     List<CostWeight> costWeights = List.of();
+
+    HasClusterCost clusterCost() {
+      if (costWeights.isEmpty()) throw new IllegalArgumentException("costWeights is not specified");
+      var config =
+          Configuration.of(
+              costWeights.stream()
+                  .collect(Collectors.toMap(e -> e.cost, e -> String.valueOf(e.weight))));
+      var fs = Utils.costFunctions(config, HasClusterCost.class);
+      if (fs.size() != costWeights.size())
+        throw new IllegalArgumentException(
+            "Has invalid costs: "
+                + costWeights.stream()
+                    .filter(
+                        e ->
+                            fs.keySet().stream()
+                                .map(c -> c.getClass().getName())
+                                .noneMatch(n -> e.cost.equals(n)))
+                    .map(e -> e.cost)
+                    .collect(Collectors.joining(",")));
+      return HasClusterCost.of(fs);
+    }
   }
 
   static class CostWeight implements Request {
-    Optional<String> cost = Optional.empty();
-    Optional<Double> weight = Optional.empty();
+    String cost;
+    double weight = 1.D;
   }
 
   static class BalancerPutRequest implements Request {
@@ -681,7 +658,7 @@ class BalancerHandler implements Handler {
     PlanExecutionProgress(
         String id,
         PlanPhase phase,
-        long timeoutMs,
+        Duration timeout,
         String balancer,
         String function,
         String exception,
@@ -690,7 +667,7 @@ class BalancerHandler implements Handler {
       this.phase = phase;
       this.exception = exception;
       this.plan = plan;
-      this.config = new PlanConfiguration(balancer, function, timeoutMs);
+      this.config = new PlanConfiguration(balancer, function, timeout);
     }
   }
 
@@ -724,12 +701,12 @@ class BalancerHandler implements Handler {
 
     final String function;
 
-    final long timeoutMs;
+    final Duration timeout;
 
-    PlanConfiguration(String balancer, String function, long timeoutMs) {
+    PlanConfiguration(String balancer, String function, Duration timeout) {
       this.balancer = balancer;
       this.function = function;
-      this.timeoutMs = timeoutMs;
+      this.timeout = timeout;
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/app/web/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
index e888f195..9f4be56e 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -221,8 +221,7 @@ public class RecordHandler implements Handler {
         Utils.packException(
             () ->
                 new PostResponse(
-                    result.get(
-                        Utils.toDuration(postRequest.timeout()).toNanos(), TimeUnit.NANOSECONDS))));
+                    result.get(postRequest.timeout().toNanos(), TimeUnit.NANOSECONDS))));
   }
 
   @Override
@@ -352,7 +351,7 @@ public class RecordHandler implements Handler {
 
   static class RecordPostRequest implements Request {
     private boolean async = false;
-    private String timeout = "5s";
+    private Duration timeout = Duration.ofSeconds(5);
     private List<PostRecord> records = List.of();
 
     private Optional<String> transactionId = Optional.empty();
@@ -363,7 +362,7 @@ public class RecordHandler implements Handler {
       return async;
     }
 
-    public String timeout() {
+    public Duration timeout() {
       return timeout;
     }
 
diff --git a/common/src/main/java/org/astraea/common/Utils.java b/common/src/main/java/org/astraea/common/Utils.java
index 18d4840c..c2f77b36 100644
--- a/common/src/main/java/org/astraea/common/Utils.java
+++ b/common/src/main/java/org/astraea/common/Utils.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
@@ -38,7 +39,9 @@ import java.util.function.Supplier;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import javax.management.InstanceNotFoundException;
+import org.astraea.common.cost.CostFunction;
 
 public final class Utils {
 
@@ -392,5 +395,30 @@ public final class Utils {
         .values();
   }
 
+  @SuppressWarnings("unchecked")
+  public static <T extends CostFunction> Map<T, Double> costFunctions(
+      Configuration config, Class<T> costClz) {
+    return config.entrySet().stream()
+        .flatMap(
+            nameAndWeight -> {
+              try {
+                var clz = Class.forName(nameAndWeight.getKey());
+                if (!costClz.isAssignableFrom(clz)) return Stream.of();
+                var weight = Double.parseDouble(nameAndWeight.getValue());
+                if (weight < 0.0)
+                  throw new IllegalArgumentException(
+                      "the weight of cost function should be bigger than zero");
+                return Stream.of(Map.entry((Class<T>) clz, weight));
+              } catch (ClassNotFoundException ignore) {
+                // this config is not cost function, so we just skip it.
+                return Stream.of();
+              } catch (NumberFormatException e) {
+                throw new IllegalArgumentException(
+                    "the weight of cost function must be positive number", e);
+              }
+            })
+        .collect(Collectors.toMap(e -> Utils.construct(e.getKey(), config), Map.Entry::getValue));
+  }
+
   private Utils() {}
 }
diff --git a/common/src/main/java/org/astraea/common/assignor/Assignor.java b/common/src/main/java/org/astraea/common/assignor/Assignor.java
index 9c99453f..42c45c69 100644
--- a/common/src/main/java/org/astraea/common/assignor/Assignor.java
+++ b/common/src/main/java/org/astraea/common/assignor/Assignor.java
@@ -20,7 +20,6 @@ import java.net.InetSocketAddress;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -115,38 +114,6 @@ public abstract class Assignor implements ConsumerPartitionAssignor, Configurabl
     }
   }
 
-  /**
-   * Parse cost function names and weight. you can specify multiple cost function with assignor. The
-   * format of key and value pair is "<CostFunction name>"="<weight>". For instance,
-   * {"org.astraea.common.cost.ReplicaSizeCost","1"} will be parsed to {(HasPartitionCost object),
-   * 1.0}.
-   *
-   * @param config the configuration of the user setting, contain cost function and its weight.
-   * @return Map from cost function object to its weight
-   */
-  static Map<HasPartitionCost, Double> parseCostFunctionWeight(Configuration config) {
-    return config.entrySet().stream()
-        .map(
-            nameAndWeight -> {
-              Class<?> clz;
-              try {
-                clz = Class.forName(nameAndWeight.getKey());
-              } catch (ClassNotFoundException ignore) {
-                return null;
-              }
-              var weight = Double.parseDouble(nameAndWeight.getValue());
-              if (weight < 0.0)
-                throw new IllegalArgumentException("Cost function weight should not be negative");
-              return Map.entry(clz, weight);
-            })
-        .filter(Objects::nonNull)
-        .filter(e -> HasPartitionCost.class.isAssignableFrom(e.getKey()))
-        .collect(
-            Collectors.toMap(
-                e -> Utils.construct((Class<HasPartitionCost>) e.getKey(), config),
-                Map.Entry::getValue));
-  }
-
   // -----------------------[kafka method]-----------------------//
 
   @Override
@@ -178,7 +145,7 @@ public abstract class Assignor implements ConsumerPartitionAssignor, Configurabl
             configs.entrySet().stream()
                 .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toString())));
     config.string(ConsumerConfigs.BOOTSTRAP_SERVERS_CONFIG).ifPresent(s -> bootstrap = s);
-    var costFunctions = parseCostFunctionWeight(config);
+    var costFunctions = Utils.costFunctions(config, HasPartitionCost.class);
     var customJMXPort = PartitionerUtils.parseIdJMXPort(config);
     var defaultJMXPort = config.integer(JMX_PORT);
     this.costFunction =
diff --git a/common/src/main/java/org/astraea/common/json/JsonConverter.java b/common/src/main/java/org/astraea/common/json/JsonConverter.java
index 01bb728d..c455d587 100644
--- a/common/src/main/java/org/astraea/common/json/JsonConverter.java
+++ b/common/src/main/java/org/astraea/common/json/JsonConverter.java
@@ -20,21 +20,34 @@ import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.fasterxml.jackson.annotation.JsonSetter;
 import com.fasterxml.jackson.annotation.Nulls;
+import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationContext;
 import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.deser.std.FromStringDeserializer;
 import com.fasterxml.jackson.databind.introspect.VisibilityChecker.Std;
 import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.ser.std.StdScalarSerializer;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+import java.io.IOException;
 import java.lang.reflect.Type;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
+import org.astraea.common.DataSize;
+import org.astraea.common.Utils;
 
 public interface JsonConverter {
 
@@ -47,10 +60,39 @@ public interface JsonConverter {
     return jackson();
   }
 
+  static <T> JsonDeserializer<T> deserializer(Class<T> clz, Function<String, T> function) {
+    return new FromStringDeserializer<>(clz) {
+      @Override
+      protected T _deserialize(String value, DeserializationContext ctxt) {
+        return function.apply(value);
+      }
+    };
+  }
+
+  static <T> JsonSerializer<T> serialzer(Class<T> clz, Function<T, String> function) {
+    return new StdScalarSerializer<>(clz) {
+      @Override
+      public void serialize(T value, JsonGenerator gen, SerializerProvider provider)
+          throws IOException {
+        gen.writeString(function.apply(value));
+      }
+    };
+  }
+
   static JsonConverter jackson() {
+    var module = new SimpleModule();
+    // DataSize
+    module.addDeserializer(DataSize.class, deserializer(DataSize.class, DataSize::of));
+    module.addSerializer(DataSize.class, serialzer(DataSize.class, DataSize::toString));
+    // Duration
+    module.addDeserializer(Duration.class, deserializer(Duration.class, Utils::toDuration));
+    // TODO: how to support ns?
+    // https://github.com/skiptests/astraea/issues/1430
+    module.addSerializer(Duration.class, serialzer(Duration.class, d -> d.toMillis() + "ms"));
     var objectMapper =
         JsonMapper.builder()
             .addModule(new Jdk8Module())
+            .addModule(module)
             .configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false)
 
             // When we put json as key into kafka, we want to pass the same json to the same
@@ -143,7 +185,9 @@ public interface JsonConverter {
         || objClass == Byte.class
         || objClass == Boolean.class
         || objClass == String.class
-        || objClass == Object.class) {
+        || objClass == Object.class
+        || objClass == DataSize.class
+        || objClass == Duration.class) {
       return;
     }
     Arrays.stream(objClass.getDeclaredFields())
@@ -151,6 +195,7 @@ public interface JsonConverter {
         .forEach(
             x -> {
               try {
+                System.out.println("name: " + x.getName() + " isSynthetic: " + x.isSynthetic());
                 preventNull(name + "." + x.getName(), x.get(obj));
               } catch (IllegalAccessException e) {
                 throw new RuntimeException(e);
diff --git a/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java b/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java
index fae404fd..9567dd68 100644
--- a/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java
+++ b/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java
@@ -19,7 +19,6 @@ package org.astraea.common.partitioner;
 import java.net.InetSocketAddress;
 import java.time.Duration;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -134,7 +133,7 @@ public class StrictCostDispatcher extends Dispatcher {
 
   @Override
   public void configure(Configuration config) {
-    var configuredFunctions = parseCostFunctionWeight(config);
+    var configuredFunctions = Utils.costFunctions(config, HasBrokerCost.class);
     if (!configuredFunctions.isEmpty()) this.costFunction = HasBrokerCost.of(configuredFunctions);
     var customJmxPort = PartitionerUtils.parseIdJMXPort(config);
     var defaultJmxPort = config.integer(JMX_PORT);
@@ -151,39 +150,6 @@ public class StrictCostDispatcher extends Dispatcher {
     this.roundRobinKeeper = RoundRobinKeeper.of(ROUND_ROBIN_LENGTH, roundRobinLease);
   }
 
-  /**
-   * Helps parse cost-function names and weights. The format of the key and value is "<CostFunction
-   * name>"="<weight>". For example, {"org.astraea.cost.broker.BrokerInputCost", "20"} will be
-   * parsed to {(BrokerInputCost object), 20.0}.
-   *
-   * @param config that contains cost-function names and its corresponding weight
-   * @return pairs of cost-function object and its corresponding weight
-   */
-  @SuppressWarnings("unchecked")
-  public static Map<HasBrokerCost, Double> parseCostFunctionWeight(Configuration config) {
-    return config.entrySet().stream()
-        .map(
-            nameAndWeight -> {
-              Class<?> clz;
-              try {
-                clz = Class.forName(nameAndWeight.getKey());
-              } catch (ClassNotFoundException ignore) {
-                // this config is not cost function, so we just skip it.
-                return null;
-              }
-              var weight = Double.parseDouble(nameAndWeight.getValue());
-              if (weight < 0.0)
-                throw new IllegalArgumentException("Cost-function weight should not be negative");
-              return Map.entry(clz, weight);
-            })
-        .filter(Objects::nonNull)
-        .filter(e -> HasBrokerCost.class.isAssignableFrom(e.getKey()))
-        .collect(
-            Collectors.toMap(
-                e -> Utils.construct((Class<HasBrokerCost>) e.getKey(), config),
-                Map.Entry::getValue));
-  }
-
   @Override
   public void close() {
     metricCollector.close();
diff --git a/docs/web_server/web_api_balancer_chinese.md b/docs/web_server/web_api_balancer_chinese.md
index 23b3d0db..9f688979 100644
--- a/docs/web_server/web_api_balancer_chinese.md
+++ b/docs/web_server/web_api_balancer_chinese.md
@@ -53,7 +53,7 @@ curl -X POST http://localhost:8001/balancer \
         { "cost": "org.astraea.common.cost.ReplicaLeaderCost", "weight": 1 }
       ],
       "maxMigratedSize": "300MB",
-      "maxMigratedLeader": "3"
+      "maxMigratedLeader": 3
     }'
 ```
 


Test patch:
diff --git a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
index ae241288..7b5312ca 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -21,7 +21,6 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
@@ -66,6 +65,7 @@ import org.astraea.common.cost.HasClusterCost;
 import org.astraea.common.cost.HasMoveCost;
 import org.astraea.common.cost.NoSufficientMetricsException;
 import org.astraea.common.json.JsonConverter;
+import org.astraea.common.json.TypeRef;
 import org.astraea.common.metrics.collector.Fetcher;
 import org.astraea.common.metrics.platform.HostMetrics;
 import org.astraea.common.metrics.platform.JvmMemory;
@@ -118,21 +118,14 @@ public class BalancerHandlerTest {
           .replicaStream()
           .forEach(r -> Assertions.assertNotEquals(0, r.size()));
       var handler = new BalancerHandler(admin);
-      var progress =
-          submitPlanGeneration(
-              handler,
-              Map.of(
-                  BALANCER_IMPLEMENTATION_KEY,
-                  GreedyBalancer.class.getName(),
-                  BALANCER_CONFIGURATION_KEY,
-                  Map.of("a", "b"),
-                  TIMEOUT_KEY,
-                  "1234ms",
-                  COST_WEIGHT_KEY,
-                  defaultDecreasing));
+      var request = new BalancerPostRequest();
+      request.balancer = GreedyBalancer.class.getName();
+      request.balancerConfig = Map.of("a", "b");
+      request.timeout = Duration.ofMillis(1234);
+      var progress = submitPlanGeneration(handler, request);
       var report = progress.plan;
       Assertions.assertNotNull(progress.id);
-      Assertions.assertEquals(1234, progress.config.timeoutMs);
+      Assertions.assertEquals(Duration.ofMillis(1234), progress.config.timeout);
       Assertions.assertEquals(GreedyBalancer.class.getName(), progress.config.balancer);
       Assertions.assertNotEquals(0, report.changes.size());
       Assertions.assertTrue(report.cost >= report.newCost.get());
@@ -169,18 +162,11 @@ public class BalancerHandlerTest {
       var handler = new BalancerHandler(admin);
       // For all 5 topics, we only allow the first two topics can be altered.
       // We apply this limitation to test if the BalancerHandler.TOPICS_KEY works correctly.
-      var allowedTopics = topicNames.subList(0, 2);
-      var report =
-          submitPlanGeneration(
-                  handler,
-                  Map.of(
-                      BALANCER_CONFIGURATION_KEY,
-                      Map.of("iteration", "30"),
-                      TOPICS_KEY,
-                      String.join(",", allowedTopics),
-                      COST_WEIGHT_KEY,
-                      defaultDecreasing))
-              .plan;
+      var allowedTopics = List.copyOf(topicNames).subList(0, 2);
+      var request = new BalancerPostRequest();
+      request.balancerConfig = Map.of("iteration", "30");
+      request.topics = Set.copyOf(allowedTopics);
+      var report = submitPlanGeneration(handler, request).plan;
       Assertions.assertTrue(
           report.changes.stream().map(x -> x.topic).allMatch(allowedTopics::contains),
           "Only allowed topics been altered");
@@ -200,17 +186,17 @@ public class BalancerHandlerTest {
     }
   }
 
-  private static List<String> createAndProduceTopic(int topicCount) {
+  private static Set<String> createAndProduceTopic(int topicCount) {
     return createAndProduceTopic(topicCount, 3, (short) 1, true);
   }
 
-  private static List<String> createAndProduceTopic(
+  private static Set<String> createAndProduceTopic(
       int topicCount, int partitions, short replicas, boolean skewed) {
     try (var admin = Admin.of(SERVICE.bootstrapServers())) {
       var topics =
           IntStream.range(0, topicCount)
               .mapToObj(ignored -> Utils.randomString(10))
-              .collect(Collectors.toUnmodifiableList());
+              .collect(Collectors.toUnmodifiableSet());
       topics.forEach(
           topic -> {
             admin
@@ -364,17 +350,10 @@ public class BalancerHandlerTest {
     createAndProduceTopic(3);
     try (var admin = Admin.of(SERVICE.bootstrapServers())) {
       var handler = new BalancerHandler(admin);
-      var report =
-          submitPlanGeneration(
-                  handler,
-                  Map.of(
-                      MAX_MIGRATE_LEADER_KEY,
-                      leaderLimit,
-                      MAX_MIGRATE_SIZE_KEY,
-                      sizeLimit,
-                      COST_WEIGHT_KEY,
-                      defaultDecreasing))
-              .plan;
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.maxMigratedSize = DataSize.of(sizeLimit);
+      request.maxMigratedLeader = Long.parseLong(leaderLimit);
+      var report = submitPlanGeneration(handler, request).plan;
       report.migrationCosts.forEach(
           migrationCost -> {
             switch (migrationCost.name) {
@@ -423,7 +402,7 @@ public class BalancerHandlerTest {
               handler.get(Channel.ofTarget(post.id)).toCompletableFuture().join());
       Assertions.assertNotNull(post.id);
       Assertions.assertEquals(post.id, progress.id);
-      Assertions.assertEquals(996, progress.config.timeoutMs);
+      Assertions.assertEquals(Duration.ofMillis(996), progress.config.timeout);
       Assertions.assertEquals(GreedyBalancer.class.getName(), progress.config.balancer);
       Assertions.assertEquals(BalancerHandler.PlanPhase.Searched, progress.phase, "search done");
       Assertions.assertNotNull(progress.exception, "hint about no plan found");
@@ -451,14 +430,9 @@ public class BalancerHandlerTest {
     try (var admin = Admin.of(SERVICE.bootstrapServers())) {
       var theExecutor = new NoOpExecutor();
       var handler = new BalancerHandler(admin, theExecutor);
-      var progress =
-          submitPlanGeneration(
-              handler,
-              Map.of(
-                  COST_WEIGHT_KEY,
-                  defaultDecreasing,
-                  BALANCER_CONFIGURATION_KEY,
-                  Map.of("iteration", "100")));
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.balancerConfig = Map.of("iteration", "100");
+      var progress = submitPlanGeneration(handler, request);
       var thePlanId = progress.id;
 
       // act
@@ -512,7 +486,7 @@ public class BalancerHandlerTest {
       Utils.sleep(Duration.ofSeconds(3));
       var theExecutor = new NoOpExecutor();
       var handler = new BalancerHandler(admin, theExecutor);
-      var progress = submitPlanGeneration(handler, Map.of());
+      var progress = submitPlanGeneration(handler, new BalancerPostRequest());
 
       // use many threads to increase the chance to trigger a data race
       final int threadCount = Runtime.getRuntime().availableProcessors() * 3;
@@ -566,8 +540,8 @@ public class BalancerHandlerTest {
             }
           };
       var handler = new BalancerHandler(admin, theExecutor);
-      var plan0 = submitPlanGeneration(handler, Map.of());
-      var plan1 = submitPlanGeneration(handler, Map.of());
+      var plan0 = submitPlanGeneration(handler, new BalancerPostRequest());
+      var plan1 = submitPlanGeneration(handler, new BalancerPostRequest());
 
       Assertions.assertDoesNotThrow(
           () -> handler.put(httpRequest(Map.of("id", plan0.id))).toCompletableFuture().join());
@@ -596,12 +570,9 @@ public class BalancerHandlerTest {
       }
 
       var handler = new BalancerHandler(admin, new NoOpExecutor());
-      var theReport =
-          submitPlanGeneration(
-              handler,
-              Map.of(
-                  TOPICS_KEY, theTopic,
-                  COST_WEIGHT_KEY, defaultDecreasing));
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.topics = Set.of(theTopic);
+      var theReport = submitPlanGeneration(handler, request);
 
       // create an ongoing reassignment
       Assertions.assertEquals(
@@ -695,16 +666,13 @@ public class BalancerHandlerTest {
   @Test
   @Timeout(value = 60)
   void testPutSanityCheck() {
-    var topic = createAndProduceTopic(1).get(0);
+    var topic = createAndProduceTopic(1).iterator().next();
     try (var admin = Admin.of(SERVICE.bootstrapServers())) {
       var theExecutor = new NoOpExecutor();
       var handler = new BalancerHandler(admin, theExecutor);
-      var theProgress =
-          submitPlanGeneration(
-              handler,
-              Map.of(
-                  COST_WEIGHT_KEY, defaultDecreasing,
-                  TOPICS_KEY, topic));
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.topics = Set.of(topic);
+      var theProgress = submitPlanGeneration(handler, request);
 
       // pick a partition and alter its placement
       var theChange = theProgress.plan.changes.stream().findAny().orElseThrow();
@@ -753,7 +721,7 @@ public class BalancerHandlerTest {
             }
           };
       var handler = new BalancerHandler(admin, theExecutor);
-      var progress = submitPlanGeneration(handler, Map.of());
+      var progress = submitPlanGeneration(handler, new BalancerPostRequest());
       Assertions.assertEquals(BalancerHandler.PlanPhase.Searched, progress.phase);
 
       // not scheduled yet
@@ -883,10 +851,9 @@ public class BalancerHandlerTest {
     var topics = createAndProduceTopic(3);
     try (var admin = Admin.of(SERVICE.bootstrapServers())) {
       var handler = new BalancerHandler(admin, new StraightPlanExecutor());
-      var progress =
-          submitPlanGeneration(
-              handler,
-              Map.of(COST_WEIGHT_KEY, defaultDecreasing, TOPICS_KEY, String.join(",", topics)));
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.topics = topics;
+      var progress = submitPlanGeneration(handler, request);
 
       Assertions.assertDoesNotThrow(
           () -> handler.put(httpRequest(Map.of("id", progress.id))).toCompletableFuture().join(),
@@ -935,18 +902,12 @@ public class BalancerHandlerTest {
             newInvoked.set(true);
           });
 
-      var progress =
-          submitPlanGeneration(
-              handler,
-              Map.of(
-                  BALANCER_IMPLEMENTATION_KEY,
-                  balancer,
-                  BALANCER_CONFIGURATION_KEY,
-                  balancerConfig,
-                  COST_WEIGHT_KEY,
-                  defaultDecreasing,
-                  TOPICS_KEY,
-                  String.join(",", topics)));
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.balancer = balancer;
+      request.balancerConfig = balancerConfig;
+      request.topics = topics;
+
+      var progress = submitPlanGeneration(handler, request);
 
       Assertions.assertEquals(BalancerHandler.PlanPhase.Searched, progress.phase, "Plan is here");
       Assertions.assertTrue(newInvoked.get(), "The customized balancer is created");
@@ -956,16 +917,24 @@ public class BalancerHandlerTest {
 
   @Test
   void testParsePostRequest() {
-    try (Admin admin = Admin.of(SERVICE.bootstrapServers())) {
+    try (var admin = Admin.of(SERVICE.bootstrapServers())) {
+      // create a topic to avoid empty cluster
+      admin
+          .creator()
+          .topic(Utils.randomString())
+          .numberOfPartitions(3)
+          .run()
+          .toCompletableFuture()
+          .join();
+      Utils.sleep(Duration.ofSeconds(2));
+
       var clusterInfo =
           admin.topicNames(false).thenCompose(admin::clusterInfo).toCompletableFuture().join();
-      {
-        // no cost weight
-        Assertions.assertThrows(
-            IllegalArgumentException.class,
-            () -> BalancerHandler.parsePostRequestWrapper(new BalancerPostRequest(), clusterInfo),
-            "CostWeights must be specified");
-      }
+      // no cost weight
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> BalancerHandler.parsePostRequestWrapper(new BalancerPostRequest(), clusterInfo),
+          "CostWeights must be specified");
       {
         // minimal
         var request = new BalancerPostRequest();
@@ -985,8 +954,8 @@ public class BalancerHandlerTest {
         var randomTopic0 = Utils.randomString();
         var randomTopic1 = Utils.randomString();
         var request = new BalancerPostRequest();
-        request.topics = Optional.of(randomTopic0 + "," + randomTopic1);
-        request.timeout = "32";
+        request.timeout = Duration.ofSeconds(32);
+        request.topics = Set.of(randomTopic0, randomTopic1);
         request.balancerConfig = Map.of("KEY", "VALUE");
         request.costWeights = List.of(costWeight(DecreasingCost.class.getName(), 1));
 
@@ -1019,14 +988,12 @@ public class BalancerHandlerTest {
                 COST_WEIGHT_KEY,
                 defaultDecreasing);
         var balancerRequest = new BalancerPostRequest();
-        balancerRequest.topics = Optional.of("");
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () -> BalancerHandler.parsePostRequestWrapper(balancerRequest, clusterInfo),
             "Empty topic filter, nothing to rebalance");
 
         var balancerRequest3 = new BalancerPostRequest();
-        balancerRequest3.timeout = "-5566";
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () -> BalancerHandler.parsePostRequestWrapper(balancerRequest3, clusterInfo),
@@ -1034,7 +1001,7 @@ public class BalancerHandlerTest {
 
         var balancerRequest4 = new BalancerPostRequest();
         var costWeight = new CostWeight();
-        costWeight.cost = Optional.of("yes");
+        costWeight.cost = "yes";
         balancerRequest4.costWeights = List.of(costWeight);
         Assertions.assertThrows(
             IllegalArgumentException.class,
@@ -1085,18 +1052,13 @@ public class BalancerHandlerTest {
             metrics.forEach(i -> Assertions.assertInstanceOf(JvmMemory.class, i));
             invoked.set(true);
           });
-      var fetcherAndCost = Collections.singleton(costWeight(FetcherAndCost.class.getName(), 1));
+      var fetcherAndCost = List.of(costWeight(FetcherAndCost.class.getName(), 1));
 
-      var progress =
-          submitPlanGeneration(
-              handler,
-              Map.of(
-                  TIMEOUT_KEY,
-                  "8",
-                  COST_WEIGHT_KEY,
-                  fetcherAndCost,
-                  TOPICS_KEY,
-                  String.join(",", topics)));
+      var request = new BalancerHandler.BalancerPostRequest();
+      request.timeout = Duration.ofSeconds(8);
+      request.costWeights = fetcherAndCost;
+      request.topics = topics;
+      var progress = submitPlanGeneration(handler, request);
 
       Assertions.assertEquals(BalancerHandler.PlanPhase.Searched, progress.phase);
       Assertions.assertTrue(invoked.get());
@@ -1245,14 +1207,14 @@ public class BalancerHandlerTest {
 
   /** Submit the plan and wait until it generated. */
   private BalancerHandler.PlanExecutionProgress submitPlanGeneration(
-      BalancerHandler handler, Map<String, Object> requestBody) {
-    if (!requestBody.containsKey(COST_WEIGHT_KEY)) {
-      requestBody = new HashMap<>(requestBody);
-      requestBody.put(COST_WEIGHT_KEY, defaultDecreasing);
-    }
+      BalancerHandler handler, BalancerPostRequest request) {
+    if (request.costWeights.isEmpty()) request.costWeights = defaultDecreasing;
     var post =
         (BalancerHandler.PostPlanResponse)
-            handler.post(httpRequest(requestBody)).toCompletableFuture().join();
+            handler
+                .post(Channel.ofRequest(JsonConverter.defaultConverter().toJson(request)))
+                .toCompletableFuture()
+                .join();
     Utils.waitFor(
         () -> {
           var progress =
@@ -1365,12 +1327,37 @@ public class BalancerHandlerTest {
 
   private static BalancerHandler.CostWeight costWeight(String cost, double weight) {
     var cw = new BalancerHandler.CostWeight();
-    cw.cost = Optional.of(cost);
-    cw.weight = Optional.of(weight);
+    cw.cost = cost;
+    cw.weight = weight;
     return cw;
   }
 
   private static Channel httpRequest(Map<String, ?> payload) {
     return Channel.ofRequest(JsonConverter.defaultConverter().toJson(payload));
   }
+
+  @Test
+  void testJsonToBalancerPostRequest() {
+    var json =
+        "{\"balancer\":\"org.astraea.common.balancer.algorithms.GreedyBalancer\", \"topics\":[\"aa\"], \"costWeights\":[{\"cost\":\"aaa\"}]}";
+    var request =
+        JsonConverter.defaultConverter().fromJson(json, TypeRef.of(BalancerPostRequest.class));
+    Assertions.assertEquals(
+        "org.astraea.common.balancer.algorithms.GreedyBalancer", request.balancer);
+    Assertions.assertNotNull(request.balancerConfig);
+    Assertions.assertNotNull(request.timeout);
+    Assertions.assertEquals(Set.of("aa"), request.topics);
+    Assertions.assertNotNull(request.maxMigratedSize);
+    Assertions.assertEquals(1, request.costWeights.size());
+    Assertions.assertEquals("aaa", request.costWeights.get(0).cost);
+    Assertions.assertEquals(1D, request.costWeights.get(0).weight);
+
+    var noCostRequest =
+        JsonConverter.defaultConverter()
+            .fromJson(
+                "{\"balancer\":\"org.astraea.common.balancer.algorithms.GreedyBalancer\"}",
+                TypeRef.of(BalancerPostRequest.class));
+
+    Assertions.assertThrows(IllegalArgumentException.class, noCostRequest::clusterCost);
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/UtilsTest.java b/common/src/test/java/org/astraea/common/UtilsTest.java
index ab893918..b07229b1 100644
--- a/common/src/test/java/org/astraea/common/UtilsTest.java
+++ b/common/src/test/java/org/astraea/common/UtilsTest.java
@@ -25,6 +25,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.common.cost.CostFunction;
+import org.astraea.common.cost.HasBrokerCost;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -221,4 +222,41 @@ public class UtilsTest {
   private static class TestBadCostFunction implements CostFunction {
     public TestBadCostFunction(int value) {}
   }
+
+  @Test
+  void testCostFunctions() {
+    var config =
+        Configuration.of(
+            Map.of(
+                "org.astraea.common.cost.BrokerInputCost",
+                "20",
+                "org.astraea.common.cost.BrokerOutputCost",
+                "1.25"));
+    var ans = Utils.costFunctions(config, HasBrokerCost.class);
+    Assertions.assertEquals(2, ans.size());
+    for (var entry : ans.entrySet()) {
+      if (entry.getKey().getClass().getName().equals("org.astraea.common.cost.BrokerInputCost")) {
+        Assertions.assertEquals(20.0, entry.getValue());
+      } else if (entry
+          .getKey()
+          .getClass()
+          .getName()
+          .equals("org.astraea.common.cost.BrokerOutputCost")) {
+        Assertions.assertEquals(1.25, entry.getValue());
+      } else {
+        Assertions.assertEquals(0.0, entry.getValue());
+      }
+    }
+
+    // test negative weight
+    var config2 =
+        Configuration.of(
+            Map.of(
+                "org.astraea.common.cost.BrokerInputCost",
+                "-20",
+                "org.astraea.common.cost.BrokerOutputCost",
+                "1.25"));
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> Utils.costFunctions(config2, HasBrokerCost.class));
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/assignor/AssignorTest.java b/common/src/test/java/org/astraea/common/assignor/AssignorTest.java
index 1b2727e6..cb24513a 100644
--- a/common/src/test/java/org/astraea/common/assignor/AssignorTest.java
+++ b/common/src/test/java/org/astraea/common/assignor/AssignorTest.java
@@ -24,7 +24,6 @@ import java.util.Map;
 import java.util.Optional;
 import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
 import org.apache.kafka.common.TopicPartition;
-import org.astraea.common.Configuration;
 import org.astraea.common.admin.NodeInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -141,21 +140,4 @@ public class AssignorTest {
     Assertions.assertEquals(1, unregisterNode.size());
     Assertions.assertEquals("192.168.103.2", unregisterNode.get(1001));
   }
-
-  @Test
-  void testParseCostFunctionWeight() {
-    var costFunction =
-        Assignor.parseCostFunctionWeight(
-            Configuration.of(Map.of("org.astraea.common.cost.ReplicaSizeCost", "100")));
-    Assertions.assertEquals(1, costFunction.size());
-    for (var e : costFunction.entrySet()) {
-      Assertions.assertEquals(
-          "org.astraea.common.cost.ReplicaSizeCost", e.getKey().getClass().getName());
-      Assertions.assertEquals(100, e.getValue());
-    }
-
-    var negativeConfig = Configuration.of(Map.of("org.astraea.common.cost.ReplicaSizeCost", "-1"));
-    Assertions.assertThrows(
-        IllegalArgumentException.class, () -> Assignor.parseCostFunctionWeight(negativeConfig));
-  }
 }
diff --git a/common/src/test/java/org/astraea/common/json/JsonConverterTest.java b/common/src/test/java/org/astraea/common/json/JsonConverterTest.java
index 338423df..9a604e5a 100644
--- a/common/src/test/java/org/astraea/common/json/JsonConverterTest.java
+++ b/common/src/test/java/org/astraea/common/json/JsonConverterTest.java
@@ -18,12 +18,15 @@ package org.astraea.common.json;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import org.astraea.common.DataSize;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -426,4 +429,78 @@ class JsonConverterTest {
     private int intValue;
     private Double doubleValue;
   }
+
+  @Test
+  void testDataSize() {
+    var v =
+        JsonConverter.defaultConverter()
+            .fromJson("{\"value\":\"30Byte\"}", TypeRef.of(TestDataSize.class));
+    assertEquals(DataSize.Byte.of(30), v.value);
+    assertEquals(List.of(), v.list);
+    assertEquals(Map.of(), v.map);
+
+    var v1 =
+        JsonConverter.defaultConverter()
+            .fromJson(
+                "{\"value\":\"31Byte\", \"list\":[\"10MB\",\"11MB\"], \"map\":{\"a\":\"5MB\"}}",
+                TypeRef.of(TestDataSize.class));
+    assertEquals(DataSize.Byte.of(31), v1.value);
+    assertEquals(List.of(DataSize.MB.of(10), DataSize.MB.of(11)), v1.list);
+    assertEquals(Map.of("a", DataSize.MB.of(5)), v1.map);
+
+    var v2 =
+        JsonConverter.defaultConverter()
+            .fromJson(JsonConverter.defaultConverter().toJson(v1), TypeRef.of(TestDataSize.class));
+    assertEquals(v1.value, v2.value);
+    assertEquals(v1.list, v2.list);
+    assertEquals(v1.map, v2.map);
+
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> JsonConverter.defaultConverter().fromJson("{}", TypeRef.of(TestDataSize.class)));
+  }
+
+  private static class TestDataSize {
+
+    DataSize value;
+    List<DataSize> list = List.of();
+
+    Map<String, DataSize> map = Map.of();
+  }
+
+  @Test
+  void testDuration() {
+    var v =
+        JsonConverter.defaultConverter()
+            .fromJson("{\"value\":\"10s\"}", TypeRef.of(TestDuration.class));
+    assertEquals(Duration.ofSeconds(10), v.value);
+    assertEquals(List.of(), v.list);
+    assertEquals(Map.of(), v.map);
+
+    var v1 =
+        JsonConverter.defaultConverter()
+            .fromJson(
+                "{\"value\":\"11s\", \"list\":[\"13s\",\"12s\"], \"map\":{\"a\":\"7s\"}}",
+                TypeRef.of(TestDuration.class));
+    assertEquals(Duration.ofSeconds(11), v1.value);
+    assertEquals(List.of(Duration.ofSeconds(13), Duration.ofSeconds(12)), v1.list);
+    assertEquals(Map.of("a", Duration.ofSeconds(7)), v1.map);
+
+    var v2 =
+        JsonConverter.defaultConverter()
+            .fromJson(JsonConverter.defaultConverter().toJson(v1), TypeRef.of(TestDuration.class));
+    assertEquals(v1.value, v2.value);
+    assertEquals(v1.list, v2.list);
+    assertEquals(v1.map, v2.map);
+
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> JsonConverter.defaultConverter().fromJson("{}", TypeRef.of(TestDuration.class)));
+  }
+
+  private static class TestDuration {
+    Duration value;
+    List<Duration> list = List.of();
+    Map<String, Duration> map = Map.of();
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java b/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java
index 0e355922..5ece38cd 100644
--- a/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java
+++ b/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java
@@ -121,44 +121,6 @@ public class StrictCostDispatcherTest {
     }
   }
 
-  @Test
-  void testParseCostFunctionWeight() {
-    var config =
-        Configuration.of(
-            Map.of(
-                "org.astraea.common.cost.BrokerInputCost",
-                "20",
-                "org.astraea.common.cost.BrokerOutputCost",
-                "1.25"));
-    var ans = StrictCostDispatcher.parseCostFunctionWeight(config);
-    Assertions.assertEquals(2, ans.size());
-    for (var entry : ans.entrySet()) {
-      if (entry.getKey().getClass().getName().equals("org.astraea.common.cost.BrokerInputCost")) {
-        Assertions.assertEquals(20.0, entry.getValue());
-      } else if (entry
-          .getKey()
-          .getClass()
-          .getName()
-          .equals("org.astraea.common.cost.BrokerOutputCost")) {
-        Assertions.assertEquals(1.25, entry.getValue());
-      } else {
-        Assertions.assertEquals(0.0, entry.getValue());
-      }
-    }
-
-    // test negative weight
-    var config2 =
-        Configuration.of(
-            Map.of(
-                "org.astraea.common.cost.BrokerInputCost",
-                "-20",
-                "org.astraea.common.cost.BrokerOutputCost",
-                "1.25"));
-    Assertions.assertThrows(
-        IllegalArgumentException.class,
-        () -> StrictCostDispatcher.parseCostFunctionWeight(config2));
-  }
-
   public static class DumbHasBrokerCost implements HasBrokerCost {
 
     @Override