Repository: skiptests/astraea (java)

Issue:
Enhance `Balancer` with retry mechanism
Context: https://github.com/skiptests/astraea/pull/955#discussion_r1004767569

未來 #955 會使 Balancer 在遇到輸入 metrics 因爲

1. metrics 數量不夠
2. metrics 品質不良沒辦法執行演算法

原因而不能執行時，主動重試 Balancer。

這個邏輯現在會暫時寫死在 `BalancerHandler` 內，未來要將這個機制移回 `Balancer` 介面內。這樣其他模組 (e.g. GUI) 也能夠重複利用這個 feature。

Pull request description:
Enhance `Balancer` with retry mechanism
resolve #1008.

* 把 `Balancer#offer()` 的 `brokerFolders` 參數移動到 `AlgorithmConfig` 裡面。
* 把 `executionTime` 參數從 `AlgorithmConfig` 移動到 `Balancer#offer` 函數簽名上。
* 實作 default interface method `Balancer#retryOffer()`,裡面會重複呼叫 `Balancer#offer`，如果遇到 `NoSufficientMetricsException` 時，在等一段時間後主動重試這個產生

這個 PR 在做 #955 在做的事情，只是從 `Balancer` 做起而非從 `BalancerHandler` 做起。這裡面還有很多要克服的問題

* 使 `BalancerHandler` 使用 `retryOffer`
* 使 `BalancerHandler` 內計算原本 cluster cost 的邏輯移動到 balancer 內部。

上述問題會在 #955 處理掉。

Patch:
diff --git a/app/src/main/java/org/astraea/app/backup/Exporter.java b/app/src/main/java/org/astraea/app/backup/Exporter.java
index 5f87ee5a..0f0a0d30 100644
--- a/app/src/main/java/org/astraea/app/backup/Exporter.java
+++ b/app/src/main/java/org/astraea/app/backup/Exporter.java
@@ -17,42 +17,86 @@
 package org.astraea.app.backup;
 
 import com.beust.jcommander.Parameter;
+import java.io.File;
 import java.nio.file.Path;
 import java.time.Duration;
+import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
 import org.astraea.common.Utils;
+import org.astraea.common.admin.TopicPartition;
 import org.astraea.common.argument.PathField;
 import org.astraea.common.argument.StringSetField;
 import org.astraea.common.backup.RecordWriter;
 import org.astraea.common.consumer.Consumer;
 import org.astraea.common.consumer.ConsumerConfigs;
 import org.astraea.common.consumer.IteratorLimit;
+import org.astraea.common.consumer.Record;
 
 public class Exporter {
 
   public static void main(String[] args) {
     var arg = Argument.parse(new Argument(), args);
-    var iter =
-        Consumer.forTopics(arg.topics)
-            .bootstrapServers(arg.bootstrapServers())
-            .config(
-                ConsumerConfigs.AUTO_OFFSET_RESET_CONFIG,
-                ConsumerConfigs.AUTO_OFFSET_RESET_EARLIEST)
-            .config(ConsumerConfigs.GROUP_ID_CONFIG, arg.group)
-            .iterator(List.of(IteratorLimit.idle(Duration.ofSeconds(3))));
     System.out.println("prepare to export data from " + arg.topics);
-    RecordWriter.write(arg.output.toFile(), (short) 0, iter);
-    System.out.println("succeed to export data from " + arg.topics);
+    var result = execute(arg);
+    System.out.println(result);
   }
 
-  private static class Argument extends org.astraea.common.argument.Argument {
+  public static Result execute(Argument argument) {
+    if (!argument.output.toFile().isDirectory())
+      throw new IllegalArgumentException("--output must be a existent folder");
+    var root = argument.output.toFile();
+    var recordCount = new HashMap<TopicPartition, Long>();
+    for (var t : argument.topics) {
+      var iter =
+          Consumer.forTopics(Set.of(t))
+              .bootstrapServers(argument.bootstrapServers())
+              .config(
+                  ConsumerConfigs.AUTO_OFFSET_RESET_CONFIG,
+                  ConsumerConfigs.AUTO_OFFSET_RESET_EARLIEST)
+              .config(ConsumerConfigs.GROUP_ID_CONFIG, argument.group)
+              .iterator(List.of(IteratorLimit.idle(Duration.ofSeconds(3))));
+      // skip empty iter to avoid creating empty file
+      if (!iter.hasNext()) continue;
+      // TODO: we should create the folder for each partition
+      var file = new File(root, t);
+      var count = new AtomicLong();
+      RecordWriter.write(
+          file,
+          (short) 0,
+          new Iterator<>() {
+            @Override
+            public boolean hasNext() {
+              return iter.hasNext();
+            }
+
+            @Override
+            public Record<byte[], byte[]> next() {
+              var record = iter.next();
+              if (record != null) {
+                count.incrementAndGet();
+                recordCount.compute(
+                    TopicPartition.of(record.topic(), record.partition()),
+                    (k, v) -> v == null ? 1 : v + 1);
+              }
+              return record;
+            }
+          });
+      System.out.println("read " + count + " records from " + t);
+    }
+    return new Result(recordCount);
+  }
+
+  static class Argument extends org.astraea.common.argument.Argument {
 
     @Parameter(
         names = {"--topics"},
         description = "List<String>: topic names to export data",
         validateWith = StringSetField.class,
-        listConverter = StringSetField.class,
+        converter = StringSetField.class,
         required = true)
     Set<String> topics;
 
@@ -69,4 +113,21 @@ public class Exporter {
             "String: the group id used by this exporter. You can run multiples exporter with same id in parallel")
     String group = Utils.randomString();
   }
+
+  public static class Result {
+    private final Map<TopicPartition, Long> recordCount;
+
+    private Result(Map<TopicPartition, Long> recordCount) {
+      this.recordCount = recordCount;
+    }
+
+    public Map<TopicPartition, Long> recordCount() {
+      return recordCount;
+    }
+
+    @Override
+    public String toString() {
+      return "Result{" + "recordCount=" + recordCount + '}';
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/backup/Importer.java b/app/src/main/java/org/astraea/app/backup/Importer.java
index 19c70a7b..d8b57531 100644
--- a/app/src/main/java/org/astraea/app/backup/Importer.java
+++ b/app/src/main/java/org/astraea/app/backup/Importer.java
@@ -17,7 +17,14 @@
 package org.astraea.app.backup;
 
 import com.beust.jcommander.Parameter;
+import java.io.File;
 import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.astraea.common.admin.TopicPartition;
 import org.astraea.common.argument.PathField;
 import org.astraea.common.backup.RecordReader;
 import org.astraea.common.producer.Producer;
@@ -27,27 +34,55 @@ public class Importer {
   public static void main(String[] args) {
     var arg = Argument.parse(new Argument(), args);
     System.out.println("prepare to import data from " + arg.input);
-    var count = 0;
-    try (var producer = Producer.of(arg.bootstrapServers())) {
-      var iter = RecordReader.read(arg.input.toFile());
-      while (iter.hasNext()) {
-        var record = iter.next();
-        if (record.key() == null && record.value() == null) continue;
-        producer
-            .sender()
-            .topic(record.topic())
-            .partition(record.partition())
-            .key(record.key())
-            .value(record.value())
-            .timestamp(record.timestamp())
-            .run();
-        count++;
+    var result = execute(arg);
+    System.out.println(result);
+  }
+
+  public static Result execute(Argument argument) {
+    if (!argument.input.toFile().isDirectory())
+      throw new IllegalArgumentException("--input must be a existent folder");
+    var files = new LinkedList<File>();
+    files.add(argument.input.toFile());
+    var recordCount = new HashMap<TopicPartition, Long>();
+    try (var producer = Producer.of(argument.bootstrapServers())) {
+      Consumer<File> process =
+          file -> {
+            var count = 0L;
+            var iter = RecordReader.read(file);
+            while (iter.hasNext()) {
+              var record = iter.next();
+              if (record.key() == null && record.value() == null) continue;
+              producer
+                  .sender()
+                  .topic(record.topic())
+                  .partition(record.partition())
+                  .key(record.key())
+                  .value(record.value())
+                  .timestamp(record.timestamp())
+                  .run();
+              recordCount.compute(
+                  TopicPartition.of(record.topic(), record.partition()),
+                  (k, v) -> v == null ? 1 : v + 1);
+              count++;
+            }
+            System.out.println("succeed to import " + count + " records from " + file);
+          };
+
+      while (true) {
+        var current = files.poll();
+        if (current == null) break;
+        if (current.isDirectory()) {
+          var fs = current.listFiles();
+          if (fs == null) continue;
+          files.addAll(Arrays.asList(fs));
+        }
+        if (current.isFile()) process.accept(current);
       }
     }
-    System.out.println("succeed to import " + count + " records from " + arg.input);
+    return new Result(recordCount);
   }
 
-  private static class Argument extends org.astraea.common.argument.Argument {
+  static class Argument extends org.astraea.common.argument.Argument {
 
     @Parameter(
         names = {"--input"},
@@ -56,4 +91,21 @@ public class Importer {
         required = true)
     Path input;
   }
+
+  public static class Result {
+    private final Map<TopicPartition, Long> recordCount;
+
+    private Result(Map<TopicPartition, Long> recordCount) {
+      this.recordCount = recordCount;
+    }
+
+    public Map<TopicPartition, Long> recordCount() {
+      return recordCount;
+    }
+
+    @Override
+    public String toString() {
+      return "Result{" + "recordCount=" + recordCount + '}';
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
index 8fb3e1f3..57430135 100644
--- a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
+++ b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
@@ -51,9 +51,10 @@ public class BalanceProcessDemo {
                   AlgorithmConfig.builder()
                       .clusterCost(new ReplicaLeaderCost())
                       .topicFilter(filter)
+                      .dataFolders(brokerFolders)
                       .config("iteration", "1000")
                       .build())
-              .offer(clusterInfo, brokerFolders);
+              .offer(clusterInfo, Duration.ofSeconds(3));
       plan.ifPresent(
           p ->
               new StraightPlanExecutor()
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 e46d5edf..2b8332fd 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -25,6 +25,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -137,20 +138,16 @@ class BalancerHandler implements Handler {
                 admin.topicNames(false).thenCompose(admin::clusterInfo),
                 admin.brokerFolders(),
                 (currentClusterInfo, brokerFolders) -> {
-                  var balancerClasspath =
-                      channel
-                          .request()
-                          .get(BALANCER_IMPLEMENTATION_KEY)
-                          .orElse(BALANCER_IMPLEMENTATION_DEFAULT);
-                  var config = parseAlgorithmConfig(channel, currentClusterInfo);
+                  var request = parsePostRequest(channel, currentClusterInfo, brokerFolders);
                   var cost =
-                      config
+                      request
+                          .algorithmConfig
                           .clusterCostFunction()
                           .clusterCost(currentClusterInfo, ClusterBean.EMPTY)
                           .value();
                   var bestPlan =
-                      Balancer.create(balancerClasspath, config)
-                          .offer(currentClusterInfo, brokerFolders);
+                      Balancer.create(request.balancerClasspath, request.algorithmConfig)
+                          .offer(currentClusterInfo, request.executionTime);
                   var changes =
                       bestPlan
                           .map(
@@ -177,7 +174,7 @@ class BalancerHandler implements Handler {
                           newPlanId,
                           cost,
                           bestPlan.map(p -> p.clusterCost().value()).orElse(null),
-                          config.clusterCostFunction().getClass().getSimpleName(),
+                          request.algorithmConfig.clusterCostFunction().getClass().getSimpleName(),
                           changes,
                           bestPlan
                               .map(
@@ -199,8 +196,12 @@ class BalancerHandler implements Handler {
   }
 
   // visible for test
-  static AlgorithmConfig parseAlgorithmConfig(
-      Channel channel, ClusterInfo<Replica> currentClusterInfo) {
+  static PostRequest parsePostRequest(
+      Channel channel,
+      ClusterInfo<Replica> currentClusterInfo,
+      Map<Integer, Set<String>> dataFolders) {
+    var balancerClasspath =
+        channel.request().get(BALANCER_IMPLEMENTATION_KEY).orElse(BALANCER_IMPLEMENTATION_DEFAULT);
     var balancerConfig =
         channel
             .request()
@@ -232,14 +233,17 @@ class BalancerHandler implements Handler {
       throw new IllegalArgumentException(
           "Illegal timeout, value should be positive integer: " + timeout.getSeconds());
 
-    return AlgorithmConfig.builder()
-        .clusterCost(clusterCostFunction)
-        .moveCost(DEFAULT_MOVE_COST_FUNCTIONS)
-        .movementConstraint(movementConstraint(channel.request().raw()))
-        .topicFilter(topics::contains)
-        .limit(timeout)
-        .config(balancerConfig)
-        .build();
+    return new PostRequest(
+        balancerClasspath,
+        timeout,
+        AlgorithmConfig.builder()
+            .clusterCost(clusterCostFunction)
+            .dataFolders(dataFolders)
+            .moveCost(DEFAULT_MOVE_COST_FUNCTIONS)
+            .movementConstraint(movementConstraint(channel.request().raw()))
+            .topicFilter(topics::contains)
+            .config(balancerConfig)
+            .build());
   }
 
   @SuppressWarnings("unchecked")
@@ -429,6 +433,18 @@ class BalancerHandler implements Handler {
               + ongoingMigration);
   }
 
+  static class PostRequest {
+    final String balancerClasspath;
+    final Duration executionTime;
+    final AlgorithmConfig algorithmConfig;
+
+    PostRequest(String balancerClasspath, Duration executionTime, AlgorithmConfig algorithmConfig) {
+      this.balancerClasspath = balancerClasspath;
+      this.executionTime = executionTime;
+      this.algorithmConfig = algorithmConfig;
+    }
+  }
+
   static class Placement {
 
     final int brokerId;
diff --git a/common/src/main/java/org/astraea/common/balancer/Balancer.java b/common/src/main/java/org/astraea/common/balancer/Balancer.java
index 52370b89..b54df12b 100644
--- a/common/src/main/java/org/astraea/common/balancer/Balancer.java
+++ b/common/src/main/java/org/astraea/common/balancer/Balancer.java
@@ -16,10 +16,9 @@
  */
 package org.astraea.common.balancer;
 
+import java.time.Duration;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import org.astraea.common.EnumInfo;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.ClusterInfo;
@@ -30,14 +29,45 @@ import org.astraea.common.balancer.algorithms.SingleStepBalancer;
 import org.astraea.common.balancer.log.ClusterLogAllocation;
 import org.astraea.common.cost.ClusterCost;
 import org.astraea.common.cost.MoveCost;
+import org.astraea.common.cost.NoSufficientMetricsException;
 
 public interface Balancer {
 
+  /**
+   * Execute {@link Balancer#offer(ClusterInfo, Duration)}. Retry the plan generation if a {@link
+   * NoSufficientMetricsException} exception occurred.
+   */
+  default Optional<Plan> retryOffer(ClusterInfo<Replica> currentClusterInfo, Duration timeout) {
+    final var timeoutMs = System.currentTimeMillis() + timeout.toMillis();
+    while (System.currentTimeMillis() < timeoutMs) {
+      try {
+        return offer(currentClusterInfo, Duration.ofMillis(timeoutMs - System.currentTimeMillis()));
+      } catch (NoSufficientMetricsException e) {
+        e.printStackTrace();
+        var remainTimeout = timeoutMs - System.currentTimeMillis();
+        var waitMs = e.suggestedWait().toMillis();
+        if (remainTimeout > waitMs) {
+          Utils.sleep(Duration.ofMillis(waitMs));
+        } else {
+          // This suggested wait time will definitely time out after we woke up
+          throw new RuntimeException(
+              "Execution time will exceeded, "
+                  + "remain: "
+                  + remainTimeout
+                  + "ms, suggestedWait: "
+                  + waitMs
+                  + "ms.",
+              e);
+        }
+      }
+    }
+    throw new RuntimeException("Execution time exceeded: " + timeoutMs);
+  }
+
   /**
    * @return a rebalance plan
    */
-  Optional<Plan> offer(
-      ClusterInfo<Replica> currentClusterInfo, Map<Integer, Set<String>> brokerFolders);
+  Optional<Plan> offer(ClusterInfo<Replica> currentClusterInfo, Duration timeout);
 
   @SuppressWarnings("unchecked")
   static Balancer create(String classpath, AlgorithmConfig config) {
diff --git a/common/src/main/java/org/astraea/common/balancer/algorithms/AlgorithmConfig.java b/common/src/main/java/org/astraea/common/balancer/algorithms/AlgorithmConfig.java
index f9720b46..bdaa3734 100644
--- a/common/src/main/java/org/astraea/common/balancer/algorithms/AlgorithmConfig.java
+++ b/common/src/main/java/org/astraea/common/balancer/algorithms/AlgorithmConfig.java
@@ -16,11 +16,11 @@
  */
 package org.astraea.common.balancer.algorithms;
 
-import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.UUID;
 import java.util.function.BiPredicate;
 import java.util.function.Predicate;
@@ -46,6 +46,11 @@ public interface AlgorithmConfig {
    */
   String executionId();
 
+  /**
+   * @return the data folders of al brokers.
+   */
+  Map<Integer, Set<String>> dataFolders();
+
   /**
    * @return the cluster cost function for this problem.
    */
@@ -66,11 +71,6 @@ public interface AlgorithmConfig {
    */
   Predicate<List<MoveCost>> movementConstraint();
 
-  /**
-   * @return the limit of algorithm execution time
-   */
-  Duration executionTime();
-
   /**
    * @return a {@link Predicate} that can indicate which topic is eligible for rebalance.
    */
@@ -90,11 +90,11 @@ public interface AlgorithmConfig {
 
     private String executionId = "noname-" + UUID.randomUUID();
     private HasClusterCost clusterCostFunction;
+    private Map<Integer, Set<String>> dataFolders;
     private List<HasMoveCost> moveCostFunction = List.of(HasMoveCost.EMPTY);
     private BiPredicate<ClusterCost, ClusterCost> clusterConstraint =
         (before, after) -> after.value() < before.value();
     private Predicate<List<MoveCost>> movementConstraint = ignore -> true;
-    private Duration executionTime = Duration.ofSeconds(3);
     private Supplier<ClusterBean> metricSource = () -> ClusterBean.EMPTY;
     private Predicate<String> topicFilter = ignore -> true;
     private final Map<String, String> config = new HashMap<>();
@@ -110,6 +110,18 @@ public interface AlgorithmConfig {
       return this;
     }
 
+    /**
+     * Specify the data folders of all brokers
+     *
+     * @return this.
+     */
+    public Builder dataFolders(Map<Integer, Set<String>> dataFolders) {
+      // TODO: Embedded these data folders information into ClusterInfo
+      //       see https://github.com/skiptests/astraea/issues/1106
+      this.dataFolders = Objects.requireNonNull(dataFolders);
+      return this;
+    }
+
     /**
      * Specify the cluster cost function to use. It implemented specific logic to evaluate if a
      * rebalance plan is worth using at certain performance/resource usage aspect
@@ -161,15 +173,6 @@ public interface AlgorithmConfig {
       return this;
     }
 
-    /**
-     * @param limit the execution time of searching best plan.
-     * @return this
-     */
-    public Builder limit(Duration limit) {
-      this.executionTime = limit;
-      return this;
-    }
-
     /**
      * Specify the source of bean metrics. The default supplier return {@link ClusterBean#EMPTY}
      * only, which means any cost function that interacts with metrics won't work. To use a cost
@@ -217,12 +220,20 @@ public interface AlgorithmConfig {
     }
 
     public AlgorithmConfig build() {
+      Objects.requireNonNull(clusterCostFunction);
+      Objects.requireNonNull(dataFolders);
+
       return new AlgorithmConfig() {
         @Override
         public String executionId() {
           return executionId;
         }
 
+        @Override
+        public Map<Integer, Set<String>> dataFolders() {
+          return dataFolders;
+        }
+
         @Override
         public HasClusterCost clusterCostFunction() {
           return clusterCostFunction;
@@ -243,11 +254,6 @@ public interface AlgorithmConfig {
           return movementConstraint;
         }
 
-        @Override
-        public Duration executionTime() {
-          return executionTime;
-        }
-
         @Override
         public Predicate<String> topicFilter() {
           return topicFilter;
diff --git a/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java b/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java
index 311b406b..c0f6adaf 100644
--- a/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java
+++ b/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java
@@ -16,7 +16,7 @@
  */
 package org.astraea.common.balancer.algorithms;
 
-import java.util.Map;
+import java.time.Duration;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
@@ -82,8 +82,7 @@ public class GreedyBalancer implements Balancer {
   }
 
   @Override
-  public Optional<Plan> offer(
-      ClusterInfo<Replica> currentClusterInfo, Map<Integer, Set<String>> brokerFolders) {
+  public Optional<Plan> offer(ClusterInfo<Replica> currentClusterInfo, Duration timeout) {
     Jmx jmx = new Jmx();
 
     final var allocationTweaker = new ShuffleTweaker(minStep, maxStep);
@@ -93,13 +92,13 @@ public class GreedyBalancer implements Balancer {
 
     final var loop = new AtomicInteger(iteration);
     final var start = System.currentTimeMillis();
-    final var executionTime = config.executionTime().toMillis();
+    final var executionTime = timeout.toMillis();
     Supplier<Boolean> moreRoom =
         () -> System.currentTimeMillis() - start < executionTime && loop.getAndDecrement() > 0;
     BiFunction<ClusterLogAllocation, ClusterCost, Optional<Balancer.Plan>> next =
         (currentAllocation, currentCost) ->
             allocationTweaker
-                .generate(brokerFolders, currentAllocation)
+                .generate(config.dataFolders(), currentAllocation)
                 .takeWhile(ignored -> moreRoom.get())
                 .map(
                     newAllocation -> {
diff --git a/common/src/main/java/org/astraea/common/balancer/algorithms/SingleStepBalancer.java b/common/src/main/java/org/astraea/common/balancer/algorithms/SingleStepBalancer.java
index c4d92ddd..6b165382 100644
--- a/common/src/main/java/org/astraea/common/balancer/algorithms/SingleStepBalancer.java
+++ b/common/src/main/java/org/astraea/common/balancer/algorithms/SingleStepBalancer.java
@@ -16,8 +16,8 @@
  */
 package org.astraea.common.balancer.algorithms;
 
+import java.time.Duration;
 import java.util.Comparator;
-import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
@@ -72,8 +72,7 @@ public class SingleStepBalancer implements Balancer {
   }
 
   @Override
-  public Optional<Balancer.Plan> offer(
-      ClusterInfo<Replica> currentClusterInfo, Map<Integer, Set<String>> brokerFolders) {
+  public Optional<Balancer.Plan> offer(ClusterInfo<Replica> currentClusterInfo, Duration timeout) {
     final var allocationTweaker = new ShuffleTweaker(minStep, maxStep);
     final var currentClusterBean = config.metricSource().get();
     final var clusterCostFunction = config.clusterCostFunction();
@@ -83,12 +82,11 @@ public class SingleStepBalancer implements Balancer {
     final var generatorClusterInfo = ClusterInfo.masked(currentClusterInfo, config.topicFilter());
 
     var start = System.currentTimeMillis();
-    var executionTime = config.executionTime().toMillis();
     return allocationTweaker
-        .generate(brokerFolders, ClusterLogAllocation.of(generatorClusterInfo))
+        .generate(config.dataFolders(), ClusterLogAllocation.of(generatorClusterInfo))
         .parallel()
         .limit(iteration)
-        .takeWhile(ignored -> System.currentTimeMillis() - start <= executionTime)
+        .takeWhile(ignored -> System.currentTimeMillis() - start <= timeout.toMillis())
         .map(
             newAllocation -> {
               var newClusterInfo = ClusterInfo.update(currentClusterInfo, newAllocation::replicas);
diff --git a/common/src/main/java/org/astraea/common/cost/CostFunction.java b/common/src/main/java/org/astraea/common/cost/CostFunction.java
index 7820ec94..1ebac97b 100644
--- a/common/src/main/java/org/astraea/common/cost/CostFunction.java
+++ b/common/src/main/java/org/astraea/common/cost/CostFunction.java
@@ -25,6 +25,13 @@ import org.astraea.common.metrics.collector.Fetcher;
  *
  * <p>Constructors in CostFunction can only take no or only one parameter {@link Configuration}, ex.
  * Constructor({@link Configuration} configuration) or Constructor()
+ *
+ * <p>A cost function might take advantage of some Mbean metrics to function. One can indicate such
+ * requirement in the {@link CostFunction#fetcher()} function. If the operation logic of
+ * CostFunction thinks the metrics on hand are insufficient or not ready to work. One can throw a
+ * {@link NoSufficientMetricsException} from the calculation logic of {@link HasBrokerCost} or
+ * {@link HasClusterCost}. This serves as a hint to the caller that it needs newer metrics and
+ * please retry later.
  */
 public interface CostFunction {
 
diff --git a/common/src/main/java/org/astraea/common/cost/NoSufficientMetricsException.java b/common/src/main/java/org/astraea/common/cost/NoSufficientMetricsException.java
new file mode 100644
index 00000000..2626151b
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/cost/NoSufficientMetricsException.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.common.cost;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Indicate that {@link CostFunction} thinks the metrics on hand cannot be processed. It probably is
+ * due to the insufficient amount of metrics or contradiction in the value of the metric. This
+ * exception serves as a hint to the caller that it needs more or newer metrics to function, please
+ * retry later.
+ */
+public class NoSufficientMetricsException extends RuntimeException {
+
+  private final CostFunction source;
+  private final Duration suggestedWait;
+
+  public NoSufficientMetricsException(CostFunction source, Duration suggestedWait) {
+    this(source, suggestedWait, "");
+  }
+
+  public NoSufficientMetricsException(CostFunction source, Duration suggestedWait, String message) {
+    super(composedMessage(source, message));
+    if (suggestedWait.isNegative() || suggestedWait.isZero())
+      throw new IllegalArgumentException(
+          "the wait time should be positive: " + suggestedWait.toMillis());
+    this.source = Objects.requireNonNull(source);
+    this.suggestedWait = Objects.requireNonNull(suggestedWait);
+  }
+
+  public NoSufficientMetricsException(
+      CostFunction source, Duration suggestedWait, String message, Throwable cause) {
+    super(composedMessage(source, message), cause);
+    if (suggestedWait.isNegative() || suggestedWait.isZero())
+      throw new IllegalArgumentException(
+          "the wait time should be positive: " + suggestedWait.toMillis());
+    this.source = Objects.requireNonNull(source);
+    this.suggestedWait = Objects.requireNonNull(suggestedWait);
+  }
+
+  /** Which cost function cause this exception. */
+  public CostFunction source() {
+    return source;
+  }
+
+  /**
+   * The suggested retry interval. This is just a suggestion so the implementation might wait longer
+   * than the specified duration.
+   */
+  public Duration suggestedWait() {
+    return suggestedWait;
+  }
+
+  private static String composedMessage(CostFunction source, String original) {
+    return "Not enough metrics for " + source.getClass().getName() + ": " + original;
+  }
+}
diff --git a/docker/start_app.sh b/docker/start_app.sh
index 59378515..d654f86b 100755
--- a/docker/start_app.sh
+++ b/docker/start_app.sh
@@ -111,7 +111,11 @@ function runContainer() {
   local defined_file="false"
   for word in "${sentence[@]}"; do
     # user has pre-defined directories/files, so we will mount directories/files
-    if [[ "$word" == "--prop.file" || "$word" == "--report.path" ]]; then
+    # --prop.file is used by all applications
+    # --report.path is used by org.astraea.app.performance.Performance
+    # --output is used by org.astraea.app.backup.Exporter
+    # --input is used by org.astraea.app.backup.Importer
+    if [[ "$word" == "--prop.file" || "$word" == "--report.path" || "$word" == "--output" || "$word" == "--input" ]]; then
       defined_file="true"
       continue
     fi
diff --git a/gui/src/main/java/org/astraea/gui/tab/BalancerNode.java b/gui/src/main/java/org/astraea/gui/tab/BalancerNode.java
index d5fd7e9a..dad1cc3c 100644
--- a/gui/src/main/java/org/astraea/gui/tab/BalancerNode.java
+++ b/gui/src/main/java/org/astraea/gui/tab/BalancerNode.java
@@ -208,13 +208,13 @@ public class BalancerNode {
                                               .clusterCost(
                                                   HasClusterCost.of(
                                                       clusterCosts(argument.selectedKeys())))
+                                              .dataFolders(brokerFolders)
                                               .moveCost(
                                                   List.of(
                                                       new ReplicaSizeCost(),
                                                       new ReplicaLeaderCost()))
                                               .movementConstraint(
                                                   movementConstraint(argument.nonEmptyTexts()))
-                                              .limit(Duration.ofSeconds(10))
                                               .topicFilter(
                                                   topic ->
                                                       patterns.isEmpty()
@@ -223,7 +223,7 @@ public class BalancerNode {
                                                                   p -> p.matcher(topic).matches()))
                                               .config("iteration", "10000")
                                               .build())
-                                      .offer(clusterInfo, brokerFolders));
+                                      .offer(clusterInfo, Duration.ofSeconds(10)));
                             }))
             .thenApply(
                 entry -> {


Test patch:
diff --git a/app/src/test/java/org/astraea/app/backup/TestImportExport.java b/app/src/test/java/org/astraea/app/backup/TestImportExport.java
new file mode 100644
index 00000000..fad28a51
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/backup/TestImportExport.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.backup;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.IntStream;
+import org.astraea.common.Utils;
+import org.astraea.common.admin.Admin;
+import org.astraea.common.argument.Argument;
+import org.astraea.common.producer.Producer;
+import org.astraea.it.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestImportExport extends RequireBrokerCluster {
+
+  @Test
+  void test() throws IOException {
+    var records = 30;
+    var topics = Set.of(Utils.randomString(), Utils.randomString(), Utils.randomString());
+    try (var producer = Producer.of(bootstrapServers())) {
+      topics.forEach(
+          t ->
+              IntStream.range(0, records)
+                  .forEach(
+                      i ->
+                          producer
+                              .sender()
+                              .topic(t)
+                              .key(Utils.randomString().getBytes(StandardCharsets.UTF_8))
+                              .run()));
+    }
+    var group = Utils.randomString();
+    var file = Files.createTempDirectory("test_import_export");
+
+    // test export
+    var exportArg =
+        Argument.parse(
+            new Exporter.Argument(),
+            new String[] {
+              "--bootstrap.servers", bootstrapServers(),
+              "--topics", String.join(",", topics),
+              "--output", file.toString(),
+              "--group", group
+            });
+    var exportResult = Exporter.execute(exportArg);
+    Assertions.assertEquals(topics.size(), exportResult.recordCount().size());
+    exportResult.recordCount().values().forEach(v -> Assertions.assertEquals(records, v));
+    // TODO: we should create the folder for each partition
+    Assertions.assertEquals(
+        topics.size(),
+        Arrays.stream(Objects.requireNonNull(file.toFile().listFiles()))
+            .filter(File::isFile)
+            .count());
+
+    // use the same group and there is no more records
+    Assertions.assertEquals(0, Exporter.execute(exportArg).recordCount().size());
+
+    // test export
+    var importArg =
+        Argument.parse(
+            new Importer.Argument(),
+            new String[] {
+              "--bootstrap.servers", bootstrapServers(),
+              "--input", file.toString()
+            });
+    var importResult = Importer.execute(importArg);
+    Assertions.assertEquals(topics.size(), importResult.recordCount().size());
+    importResult.recordCount().values().forEach(v -> Assertions.assertEquals(records, v));
+    try (var admin = Admin.of(bootstrapServers())) {
+      var offsets =
+          admin
+              .latestOffsets(admin.topicPartitions(topics).toCompletableFuture().join())
+              .toCompletableFuture()
+              .join();
+      Assertions.assertEquals(topics.size(), offsets.size());
+      offsets.values().forEach(v -> Assertions.assertEquals(records * 2, v));
+    }
+  }
+}
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 55ee1d70..92a76fcd 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -280,6 +280,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   AlgorithmConfig.builder()
                       .clusterCost(clusterCostFunction)
                       .clusterConstraint((before, after) -> after.value() <= before.value())
+                      .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                       .moveCost(List.of(moveCostFunction))
                       .movementConstraint(moveCosts -> true)
                       .build())
@@ -288,7 +289,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                       .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                       .toCompletableFuture()
                       .join(),
-                  admin.brokerFolders().toCompletableFuture().join());
+                  Duration.ofSeconds(3));
 
       Assertions.assertNotEquals(Optional.empty(), Best);
 
@@ -301,6 +302,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                       AlgorithmConfig.builder()
                           .clusterCost(clusterCostFunction)
                           .clusterConstraint((before, after) -> true)
+                          .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                           .moveCost(List.of(moveCostFunction))
                           .movementConstraint(moveCosts -> true)
                           .config("iteration", "0")
@@ -310,7 +312,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                           .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                           .toCompletableFuture()
                           .join(),
-                      admin.brokerFolders().toCompletableFuture().join()));
+                      Duration.ofSeconds(3)));
 
       // test cluster cost predicate
       Assertions.assertEquals(
@@ -320,6 +322,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   AlgorithmConfig.builder()
                       .clusterCost(clusterCostFunction)
                       .clusterConstraint((before, after) -> false)
+                      .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                       .moveCost(List.of(moveCostFunction))
                       .movementConstraint(moveCosts -> true)
                       .build())
@@ -328,7 +331,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                       .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                       .toCompletableFuture()
                       .join(),
-                  admin.brokerFolders().toCompletableFuture().join()));
+                  Duration.ofSeconds(3)));
 
       // test move cost predicate
       Assertions.assertEquals(
@@ -338,6 +341,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   AlgorithmConfig.builder()
                       .clusterCost(clusterCostFunction)
                       .clusterConstraint((before, after) -> true)
+                      .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                       .moveCost(List.of(moveCostFunction))
                       .movementConstraint(moveCosts -> false)
                       .build())
@@ -346,7 +350,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                       .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                       .toCompletableFuture()
                       .join(),
-                  admin.brokerFolders().toCompletableFuture().join()));
+                  Duration.ofSeconds(3)));
     }
   }
 
@@ -913,21 +917,24 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testParseAlgorithmConfig() {
+  void testParsePostRequest() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var clusterInfo =
           admin.topicNames(false).thenCompose(admin::clusterInfo).toCompletableFuture().join();
       {
         // default
-        var config = BalancerHandler.parseAlgorithmConfig(Channel.EMPTY, clusterInfo);
-        Assertions.assertTrue(config.algorithmConfig().entrySet().isEmpty());
-        Assertions.assertInstanceOf(HasClusterCost.class, config.clusterCostFunction());
+        var postRequest = BalancerHandler.parsePostRequest(Channel.EMPTY, clusterInfo, Map.of());
+        Assertions.assertTrue(postRequest.algorithmConfig.algorithmConfig().entrySet().isEmpty());
+        Assertions.assertInstanceOf(
+            HasClusterCost.class, postRequest.algorithmConfig.clusterCostFunction());
         Assertions.assertEquals(
-            BalancerHandler.DEFAULT_CLUSTER_COST_FUNCTION, config.clusterCostFunction());
+            BalancerHandler.DEFAULT_CLUSTER_COST_FUNCTION,
+            postRequest.algorithmConfig.clusterCostFunction());
         Assertions.assertEquals(
-            BalancerHandler.TIMEOUT_DEFAULT, config.executionTime().toSeconds());
+            BalancerHandler.TIMEOUT_DEFAULT, postRequest.executionTime.toSeconds());
         Assertions.assertTrue(
-            clusterInfo.topics().stream().allMatch(t -> config.topicFilter().test(t)));
+            clusterInfo.topics().stream()
+                .allMatch(t -> postRequest.algorithmConfig.topicFilter().test(t)));
       }
       {
         // use custom filter/timeout/balancer config/cost function
@@ -943,23 +950,41 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                 "{\"KEY\":\"VALUE\"}",
                 BalancerHandler.COST_WEIGHT_KEY,
                 defaultDecreasing);
-        var config =
-            BalancerHandler.parseAlgorithmConfig(
-                Channel.ofRequest(PostRequest.of(request)), clusterInfo);
+        var postRequest =
+            BalancerHandler.parsePostRequest(
+                Channel.ofRequest(PostRequest.of(request)), clusterInfo, Map.of());
         Assertions.assertEquals(
-            Set.of(Map.entry("KEY", "VALUE")), config.algorithmConfig().entrySet());
-        Assertions.assertInstanceOf(HasClusterCost.class, config.clusterCostFunction());
+            Set.of(Map.entry("KEY", "VALUE")),
+            postRequest.algorithmConfig.algorithmConfig().entrySet());
+        Assertions.assertInstanceOf(
+            HasClusterCost.class, postRequest.algorithmConfig.clusterCostFunction());
         Assertions.assertEquals(
-            1.0, config.clusterCostFunction().clusterCost(clusterInfo, ClusterBean.EMPTY).value());
+            1.0,
+            postRequest
+                .algorithmConfig
+                .clusterCostFunction()
+                .clusterCost(clusterInfo, ClusterBean.EMPTY)
+                .value());
         Assertions.assertEquals(
-            1.0, config.clusterCostFunction().clusterCost(clusterInfo, ClusterBean.EMPTY).value());
+            1.0,
+            postRequest
+                .algorithmConfig
+                .clusterCostFunction()
+                .clusterCost(clusterInfo, ClusterBean.EMPTY)
+                .value());
         Assertions.assertEquals(
-            1.0, config.clusterCostFunction().clusterCost(clusterInfo, ClusterBean.EMPTY).value());
-        Assertions.assertEquals(32, config.executionTime().toSeconds());
-        Assertions.assertTrue(config.topicFilter().test(randomTopic0));
-        Assertions.assertTrue(config.topicFilter().test(randomTopic1));
+            1.0,
+            postRequest
+                .algorithmConfig
+                .clusterCostFunction()
+                .clusterCost(clusterInfo, ClusterBean.EMPTY)
+                .value());
+        Assertions.assertEquals(32, postRequest.executionTime.toSeconds());
+        Assertions.assertTrue(postRequest.algorithmConfig.topicFilter().test(randomTopic0));
+        Assertions.assertTrue(postRequest.algorithmConfig.topicFilter().test(randomTopic1));
         Assertions.assertTrue(
-            clusterInfo.topics().stream().noneMatch(t -> config.topicFilter().test(t)));
+            clusterInfo.topics().stream()
+                .noneMatch(t -> postRequest.algorithmConfig.topicFilter().test(t)));
       }
       {
         // malformed content
@@ -976,52 +1001,58 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () ->
-                BalancerHandler.parseAlgorithmConfig(
+                BalancerHandler.parsePostRequest(
                     Channel.ofRequest(PostRequest.of(Map.of(BalancerHandler.TOPICS_KEY, ""))),
-                    clusterInfo),
+                    clusterInfo,
+                    Map.of()),
             "Empty topic filter, nothing to rebalance");
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () ->
-                BalancerHandler.parseAlgorithmConfig(
+                BalancerHandler.parsePostRequest(
                     Channel.ofRequest(PostRequest.of(Map.of(BalancerHandler.TIMEOUT_KEY, 0))),
-                    clusterInfo),
+                    clusterInfo,
+                    Map.of()),
             "Zero timeout");
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () ->
-                BalancerHandler.parseAlgorithmConfig(
+                BalancerHandler.parsePostRequest(
                     Channel.ofRequest(PostRequest.of(Map.of(BalancerHandler.TIMEOUT_KEY, -5566))),
-                    clusterInfo),
+                    clusterInfo,
+                    Map.of()),
             "Negative timeout");
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () ->
-                BalancerHandler.parseAlgorithmConfig(
+                BalancerHandler.parsePostRequest(
                     Channel.ofRequest(
                         PostRequest.of(
                             Map.of(BalancerHandler.COST_WEIGHT_KEY, "[{\"cost\": \"yes\"}]"))),
-                    clusterInfo),
+                    clusterInfo,
+                    Map.of()),
             "Malformed cost weight");
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () ->
-                BalancerHandler.parseAlgorithmConfig(
+                BalancerHandler.parsePostRequest(
                     Channel.ofRequest(
                         PostRequest.of(
                             Map.of(
                                 BalancerHandler.COST_WEIGHT_KEY,
                                 "[{\"cost\": \"yes\", \"weight\": \"a lot\"}]"))),
-                    clusterInfo),
+                    clusterInfo,
+                    Map.of()),
             "Malformed cost weight");
         Assertions.assertThrows(
             IllegalArgumentException.class,
             () ->
-                BalancerHandler.parseAlgorithmConfig(
+                BalancerHandler.parsePostRequest(
                     Channel.ofRequest(
                         PostRequest.of(
                             Map.of(BalancerHandler.COST_WEIGHT_KEY, "[{\"weight\": \"a lot\"}]"))),
-                    clusterInfo),
+                    clusterInfo,
+                    Map.of()),
             "Malformed cost weight");
       }
     }
@@ -1096,11 +1127,10 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
     }
 
     @Override
-    public Optional<Plan> offer(
-        ClusterInfo<Replica> currentClusterInfo, Map<Integer, Set<String>> brokerFolders) {
+    public Optional<Plan> offer(ClusterInfo<Replica> currentClusterInfo, Duration timeout) {
       offerCallbacks.forEach(Runnable::run);
       offerCallbacks.clear();
-      return super.offer(currentClusterInfo, brokerFolders);
+      return super.offer(currentClusterInfo, timeout);
     }
   }
 }
diff --git a/common/src/test/java/org/astraea/common/balancer/BalancerAlgorithmTest.java b/common/src/test/java/org/astraea/common/balancer/BalancerAlgorithmTest.java
index cae9078a..8d72922d 100644
--- a/common/src/test/java/org/astraea/common/balancer/BalancerAlgorithmTest.java
+++ b/common/src/test/java/org/astraea/common/balancer/BalancerAlgorithmTest.java
@@ -88,14 +88,14 @@ public class BalancerAlgorithmTest extends RequireBrokerCluster {
                   GreedyBalancer.class,
                   AlgorithmConfig.builder()
                       .clusterCost(new ReplicaNumberCost())
-                      .limit(Duration.ofSeconds(5))
+                      .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                       .build())
               .offer(
                   admin
                       .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                       .toCompletableFuture()
                       .join(),
-                  admin.brokerFolders().toCompletableFuture().join())
+                  Duration.ofSeconds(5))
               .get();
 
       var plan =
@@ -103,14 +103,14 @@ public class BalancerAlgorithmTest extends RequireBrokerCluster {
                   SingleStepBalancer.class,
                   AlgorithmConfig.builder()
                       .clusterCost(new ReplicaNumberCost())
-                      .limit(Duration.ofSeconds(5))
+                      .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                       .build())
               .offer(
                   admin
                       .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                       .toCompletableFuture()
                       .join(),
-                  admin.brokerFolders().toCompletableFuture().join())
+                  Duration.ofSeconds(5))
               .get();
 
       Assertions.assertTrue(plan.clusterCost.value() > planOfGreedy.clusterCost.value());
diff --git a/common/src/test/java/org/astraea/common/balancer/BalancerTest.java b/common/src/test/java/org/astraea/common/balancer/BalancerTest.java
index 54b2dc7a..5f6732fb 100644
--- a/common/src/test/java/org/astraea/common/balancer/BalancerTest.java
+++ b/common/src/test/java/org/astraea/common/balancer/BalancerTest.java
@@ -19,6 +19,7 @@ package org.astraea.common.balancer;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -35,14 +36,19 @@ import org.astraea.common.balancer.algorithms.AlgorithmConfig;
 import org.astraea.common.balancer.algorithms.GreedyBalancer;
 import org.astraea.common.balancer.algorithms.SingleStepBalancer;
 import org.astraea.common.balancer.executor.StraightPlanExecutor;
+import org.astraea.common.balancer.log.ClusterLogAllocation;
 import org.astraea.common.cost.ClusterCost;
+import org.astraea.common.cost.DecreasingCost;
 import org.astraea.common.cost.HasClusterCost;
+import org.astraea.common.cost.NoSufficientMetricsException;
 import org.astraea.common.cost.ReplicaLeaderCost;
 import org.astraea.common.metrics.BeanObject;
 import org.astraea.common.metrics.HasBeanObject;
 import org.astraea.common.scenario.Scenario;
 import org.astraea.it.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mockito;
@@ -91,15 +97,15 @@ class BalancerTest extends RequireBrokerCluster {
                   theClass,
                   AlgorithmConfig.builder()
                       .clusterCost(new ReplicaLeaderCost())
+                      .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                       .topicFilter(topic -> topic.equals(topicName))
-                      .limit(Duration.ofSeconds(10))
                       .build())
               .offer(
                   admin
                       .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                       .toCompletableFuture()
                       .join(),
-                  admin.brokerFolders().toCompletableFuture().join())
+                  Duration.ofSeconds(10))
               .orElseThrow();
       new StraightPlanExecutor()
           .run(admin, plan.proposal(), Duration.ofSeconds(10))
@@ -150,10 +156,11 @@ class BalancerTest extends RequireBrokerCluster {
                   theClass,
                   AlgorithmConfig.builder()
                       .topicFilter(t -> t.equals(theTopic))
+                      .dataFolders(brokerFolders)
                       .clusterCost(randomScore)
                       .config("iteration", "500")
                       .build())
-              .offer(clusterInfo, brokerFolders)
+              .offer(clusterInfo, Duration.ofSeconds(3))
               .get()
               .proposal();
 
@@ -191,14 +198,14 @@ class BalancerTest extends RequireBrokerCluster {
                           theClass,
                           AlgorithmConfig.builder()
                               .clusterCost((clusterInfo, bean) -> Math::random)
-                              .limit(Duration.ofSeconds(3))
+                              .dataFolders(admin.brokerFolders().toCompletableFuture().join())
                               .build())
                       .offer(
                           admin
                               .clusterInfo(admin.topicNames(false).toCompletableFuture().join())
                               .toCompletableFuture()
                               .join(),
-                          admin.brokerFolders().toCompletableFuture().join())
+                          Duration.ofSeconds(3))
                       .get()
                       .proposal());
       Utils.sleep(Duration.ofMillis(1000));
@@ -247,10 +254,11 @@ class BalancerTest extends RequireBrokerCluster {
                   theClass,
                   AlgorithmConfig.builder()
                       .clusterCost(theCostFunction)
+                      .dataFolders(Map.of())
                       .metricSource(metricSource)
                       .config("iteration", "500")
                       .build())
-              .offer(ClusterInfo.empty(), Map.of());
+              .offer(ClusterInfo.empty(), Duration.ofSeconds(3));
           Assertions.assertTrue(called.get(), "The cost function has been invoked");
         };
 
@@ -265,4 +273,52 @@ class BalancerTest extends RequireBrokerCluster {
     test.accept(8L);
     test.accept(9L);
   }
+
+  @ParameterizedTest
+  @ValueSource(ints = {1000, 2000, 3000})
+  @Timeout(10)
+  void testRetryOffer(int sampleTimeMs) {
+    var startMs = System.currentTimeMillis();
+    var costFunction = new DecreasingCost(null);
+    var fake = FakeClusterInfo.of(3, 3, 3, 3);
+    var balancer =
+        new Balancer() {
+          @Override
+          public Optional<Plan> offer(ClusterInfo<Replica> currentClusterInfo, Duration timeout) {
+            if (System.currentTimeMillis() - startMs < sampleTimeMs)
+              throw new NoSufficientMetricsException(
+                  costFunction,
+                  Duration.ofMillis(sampleTimeMs - (System.currentTimeMillis() - startMs)));
+            return Optional.of(
+                new Plan(ClusterLogAllocation.of(currentClusterInfo), () -> 0, List.of()));
+          }
+        };
+
+    Assertions.assertDoesNotThrow(() -> balancer.retryOffer(fake, Duration.ofSeconds(10)));
+    var endMs = System.currentTimeMillis();
+
+    Assertions.assertTrue(
+        sampleTimeMs < (endMs - startMs) && (endMs - startMs) < sampleTimeMs + 1000,
+        "Finished on time");
+  }
+
+  @Test
+  @Timeout(1)
+  void testRetryOfferTimeout() {
+    var timeout = Duration.ofMillis(100);
+    var costFunction = new DecreasingCost(null);
+    var fake = FakeClusterInfo.of(3, 3, 3, 3);
+    var balancer =
+        new Balancer() {
+          @Override
+          public Optional<Plan> offer(ClusterInfo<Replica> currentClusterInfo, Duration timeout) {
+            throw new NoSufficientMetricsException(
+                costFunction, Duration.ofSeconds(999), "This will takes forever");
+          }
+        };
+
+    // start
+    Assertions.assertThrows(RuntimeException.class, () -> balancer.retryOffer(fake, timeout))
+        .printStackTrace();
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/balancer/algorithms/GreedyBalancerTest.java b/common/src/test/java/org/astraea/common/balancer/algorithms/GreedyBalancerTest.java
index e3d0ac64..76337b7b 100644
--- a/common/src/test/java/org/astraea/common/balancer/algorithms/GreedyBalancerTest.java
+++ b/common/src/test/java/org/astraea/common/balancer/algorithms/GreedyBalancerTest.java
@@ -54,22 +54,22 @@ class GreedyBalancerTest {
   void testJmx() {
     var cost = new DecreasingCost(Configuration.of(Map.of()));
     var id = "TestJmx-" + UUID.randomUUID();
+    var clusterInfo = FakeClusterInfo.of(5, 5, 5, 2);
     var balancer =
         Balancer.create(
             GreedyBalancer.class,
             AlgorithmConfig.builder()
                 .executionId(id)
+                .dataFolders(clusterInfo.dataDirectories())
                 .clusterCost(cost)
-                .limit(Duration.ofMillis(300))
                 .config(GreedyBalancer.ITERATION_CONFIG, "100")
                 .build());
-    var clusterInfo = FakeClusterInfo.of(5, 5, 5, 2);
 
     try (MBeanClient client = MBeanClient.local()) {
       IntStream.range(0, 10)
           .forEach(
               run -> {
-                var plan = balancer.offer(clusterInfo, clusterInfo.dataDirectories());
+                var plan = balancer.offer(clusterInfo, Duration.ofMillis(300));
                 Assertions.assertTrue(plan.isPresent());
                 var bean =
                     Assertions.assertDoesNotThrow(