Repository: skiptests/astraea (java)

Issue:
Remove preferredLeaderElection from Balancer
這是一個討論用的議題：

我們現在把"運行 preferred replicas election"綁定在balancer上面，但是該行為必須等到 preferred replicas 完成同步後才可以執行，這個要求會導致執行 balance plan 卡在一個”已知但未定“的時間上，這會導致後續的流程出狀況

或許，我們並不需要手動去呼叫選舉，kafka 本來就有執行緒會定時巡邏並觸發 leader election，我們可以把這個工作留給 kafka 就好

Pull request:
Enable RebalancePlanExecutor to execute plan with/without sycn waiting
fix #1055 

現在`RebalancePlanExecutor`預設需要實作的部分只要完成搬移就好，最後集體的資料同步和re-election將會由預設實作完成。同時使用者可以選擇是否要等待最後的同步

Patch:
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 681c68a1..8f0c2209 100644
--- a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
+++ b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
@@ -17,6 +17,7 @@
 package org.astraea.app.balancer;
 
 import com.beust.jcommander.Parameter;
+import java.time.Duration;
 import java.util.Set;
 import java.util.function.Predicate;
 import org.astraea.common.admin.Admin;
@@ -56,7 +57,7 @@ public class BalanceProcessDemo {
       plan.ifPresent(
           p ->
               new StraightPlanExecutor()
-                  .run(admin, p.proposal().rebalancePlan())
+                  .run(admin, p.proposal().rebalancePlan(), Duration.ofHours(1))
                   .toCompletableFuture()
                   .join());
     }
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 a51e332a..57e9c29a 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -276,8 +276,9 @@ class BalancerHandler implements Handler {
                     p -> {
                       executedPlans.put(
                           thePlanId,
-                          CompletableFuture.runAsync(
-                              () -> executor.run(admin, p.proposal().rebalancePlan())));
+                          executor
+                              .run(admin, p.proposal().rebalancePlan(), Duration.ofHours(1))
+                              .toCompletableFuture());
                       lastExecutionId.set(thePlanId);
                     });
                 return new PutPlanResponse(thePlanId);
diff --git a/common/src/main/java/org/astraea/common/Utils.java b/common/src/main/java/org/astraea/common/Utils.java
index 8feff489..cf94fc72 100644
--- a/common/src/main/java/org/astraea/common/Utils.java
+++ b/common/src/main/java/org/astraea/common/Utils.java
@@ -21,9 +21,9 @@ import java.io.StringWriter;
 import java.net.InetAddress;
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Objects;
-import java.util.Set;
-import java.util.TreeSet;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -250,13 +250,13 @@ public final class Utils {
     throw new RuntimeException(attribute + " is not existent in " + object.getClass().getName());
   }
 
-  public static Set<String> constants(Class<?> clz, Predicate<String> variableNameFilter) {
+  public static List<String> constants(Class<?> clz, Predicate<String> variableNameFilter) {
     return Arrays.stream(clz.getFields())
         .filter(field -> variableNameFilter.test(field.getName()))
         .map(field -> packException(() -> field.get(null)))
         .filter(obj -> obj instanceof String)
         .map(obj -> (String) obj)
-        .collect(Collectors.toCollection(TreeSet::new));
+        .collect(Collectors.toCollection(LinkedList::new));
   }
 
   public static String toString(Throwable e) {
diff --git a/common/src/main/java/org/astraea/common/admin/BrokerConfigs.java b/common/src/main/java/org/astraea/common/admin/BrokerConfigs.java
index 5aa68ef4..7ae9f9ab 100644
--- a/common/src/main/java/org/astraea/common/admin/BrokerConfigs.java
+++ b/common/src/main/java/org/astraea/common/admin/BrokerConfigs.java
@@ -17,6 +17,7 @@
 package org.astraea.common.admin;
 
 import java.util.Set;
+import java.util.TreeSet;
 import org.astraea.common.Utils;
 
 public final class BrokerConfigs {
@@ -82,7 +83,7 @@ public final class BrokerConfigs {
   public static final String REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG =
       "replica.alter.log.dirs.io.max.bytes.per.second";
   public static final Set<String> DYNAMICAL_CONFIGS =
-      Utils.constants(BrokerConfigs.class, name -> name.endsWith("CONFIG"));
+      new TreeSet<>(Utils.constants(BrokerConfigs.class, name -> name.endsWith("CONFIG")));
   // ---------------------------------[values]---------------------------------//
 
   private BrokerConfigs() {}
diff --git a/common/src/main/java/org/astraea/common/admin/ClusterInfo.java b/common/src/main/java/org/astraea/common/admin/ClusterInfo.java
index 59e82f4e..b24e80f8 100644
--- a/common/src/main/java/org/astraea/common/admin/ClusterInfo.java
+++ b/common/src/main/java/org/astraea/common/admin/ClusterInfo.java
@@ -372,6 +372,12 @@ public interface ClusterInfo<T extends ReplicaInfo> {
     return replicaStream().map(ReplicaInfo::topicPartition).collect(Collectors.toUnmodifiableSet());
   }
 
+  default Set<TopicPartitionReplica> topicPartitionReplicas() {
+    return replicaStream()
+        .map(ReplicaInfo::topicPartitionReplica)
+        .collect(Collectors.toUnmodifiableSet());
+  }
+
   /**
    * find the node associated to node id.
    *
@@ -508,6 +514,11 @@ public interface ClusterInfo<T extends ReplicaInfo> {
       return byPartition.get().keySet();
     }
 
+    @Override
+    public Set<TopicPartitionReplica> topicPartitionReplicas() {
+      return byReplica.get().keySet();
+    }
+
     @Override
     public Set<String> topics() {
       return byTopic.get().keySet();
diff --git a/common/src/main/java/org/astraea/common/admin/TopicConfigs.java b/common/src/main/java/org/astraea/common/admin/TopicConfigs.java
index cdc28efa..903d6a2b 100644
--- a/common/src/main/java/org/astraea/common/admin/TopicConfigs.java
+++ b/common/src/main/java/org/astraea/common/admin/TopicConfigs.java
@@ -17,6 +17,7 @@
 package org.astraea.common.admin;
 
 import java.util.Set;
+import java.util.TreeSet;
 import org.astraea.common.Utils;
 
 public final class TopicConfigs {
@@ -57,7 +58,7 @@ public final class TopicConfigs {
       "follower.replication.throttled.replicas";
 
   public static final Set<String> ALL_CONFIGS =
-      Utils.constants(TopicConfigs.class, name -> name.endsWith("CONFIG"));
+      new TreeSet<>(Utils.constants(TopicConfigs.class, name -> name.endsWith("CONFIG")));
 
   // ---------------------------------[values]---------------------------------//
   public static final String CLEANUP_POLICY_COMPACT = "compact";
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 8c312e95..b521615a 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
@@ -19,6 +19,7 @@ package org.astraea.common.balancer.algorithms;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
@@ -44,7 +45,7 @@ public class GreedyBalancer implements Balancer {
       "shuffle.plan.generator.max.step";
   public static final String ITERATION_CONFIG = "iteration";
   public static final Set<String> ALL_CONFIGS =
-      Utils.constants(GreedyBalancer.class, name -> name.endsWith("CONFIG"));
+      new TreeSet<>(Utils.constants(GreedyBalancer.class, name -> name.endsWith("CONFIG")));
 
   private final AlgorithmConfig config;
   private final int minStep;
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 dda42f1b..a7966002 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
@@ -20,6 +20,7 @@ import java.util.Comparator;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.stream.Collectors;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.ClusterInfo;
@@ -37,7 +38,7 @@ public class SingleStepBalancer implements Balancer {
       "shuffle.plan.generator.max.step";
   public static final String ITERATION_CONFIG = "iteration";
   public static final Set<String> ALL_CONFIGS =
-      Utils.constants(SingleStepBalancer.class, name -> name.endsWith("CONFIG"));
+      new TreeSet<>(Utils.constants(SingleStepBalancer.class, name -> name.endsWith("CONFIG")));
 
   private final AlgorithmConfig config;
   private final int minStep;
diff --git a/common/src/main/java/org/astraea/common/balancer/executor/RebalancePlanExecutor.java b/common/src/main/java/org/astraea/common/balancer/executor/RebalancePlanExecutor.java
index d407d0a2..a2ce81b6 100644
--- a/common/src/main/java/org/astraea/common/balancer/executor/RebalancePlanExecutor.java
+++ b/common/src/main/java/org/astraea/common/balancer/executor/RebalancePlanExecutor.java
@@ -16,13 +16,30 @@
  */
 package org.astraea.common.balancer.executor;
 
+import java.time.Duration;
 import java.util.concurrent.CompletionStage;
 import org.astraea.common.admin.Admin;
-import org.astraea.common.balancer.log.ClusterLogAllocation;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.admin.Replica;
 
 /** This class associate with the logic of fulfill given rebalance plan. */
 public interface RebalancePlanExecutor {
 
-  /** This method responsible for fulfill a rebalance plan. */
-  CompletionStage<Void> run(Admin admin, ClusterLogAllocation targetAllocation);
+  static RebalancePlanExecutor of() {
+    return new StraightPlanExecutor();
+  }
+
+  /**
+   * submit the migration request to servers. It makes sure all migrated replicas get synced and all
+   * new leaders get ready. The timeout to wait sync and re-election can be "large" and
+   * "unpredictable", so it is up to caller to give a "suitable" timeout.
+   *
+   * @param admin to process request
+   * @param targetAllocation the expected assignments
+   * @param timeout to wait metadata sync for each phase. The plan could be divided into many
+   *     requests. This timeout could stop the infinite waiting for Kafka metadata. It causes
+   *     IllegalStateException if the metadata can't get synced and timeout is expired.
+   * @return a background running thread
+   */
+  CompletionStage<Void> run(Admin admin, ClusterInfo<Replica> targetAllocation, Duration timeout);
 }
diff --git a/common/src/main/java/org/astraea/common/balancer/executor/StraightPlanExecutor.java b/common/src/main/java/org/astraea/common/balancer/executor/StraightPlanExecutor.java
index 3720e341..d5d05f4b 100644
--- a/common/src/main/java/org/astraea/common/balancer/executor/StraightPlanExecutor.java
+++ b/common/src/main/java/org/astraea/common/balancer/executor/StraightPlanExecutor.java
@@ -19,17 +19,13 @@ package org.astraea.common.balancer.executor;
 import static org.astraea.common.admin.ClusterInfo.findNonFulfilledAllocation;
 
 import java.time.Duration;
-import java.time.temporal.ChronoUnit;
 import java.util.Comparator;
-import java.util.List;
 import java.util.concurrent.CompletionStage;
 import java.util.stream.Collectors;
 import org.astraea.common.admin.Admin;
+import org.astraea.common.admin.ClusterInfo;
 import org.astraea.common.admin.Replica;
 import org.astraea.common.admin.ReplicaInfo;
-import org.astraea.common.admin.TopicPartition;
-import org.astraea.common.admin.TopicPartitionReplica;
-import org.astraea.common.balancer.log.ClusterLogAllocation;
 
 /** Execute every possible migration immediately. */
 public class StraightPlanExecutor implements RebalancePlanExecutor {
@@ -37,7 +33,8 @@ public class StraightPlanExecutor implements RebalancePlanExecutor {
   public StraightPlanExecutor() {}
 
   @Override
-  public CompletionStage<Void> run(Admin admin, ClusterLogAllocation logAllocation) {
+  public CompletionStage<Void> run(
+      Admin admin, ClusterInfo<Replica> logAllocation, Duration timeout) {
     return admin
         .topicNames(true)
         .thenCompose(admin::clusterInfo)
@@ -46,81 +43,65 @@ public class StraightPlanExecutor implements RebalancePlanExecutor {
             tps ->
                 tps.stream()
                     .flatMap(tp -> logAllocation.replicas(tp).stream())
-                    .sorted(Comparator.comparing(Replica::isPreferredLeader))
                     .collect(Collectors.toList()))
-
-        // step.1 move replica to specify brokers/folders
+        // step 1: move replicas to specify brokers
         .thenCompose(
             replicas ->
-                run(admin, replicas)
-                    .thenApply(
-                        ignored ->
-                            replicas.stream()
-                                .map(Replica::topicPartitionReplica)
-                                .collect(Collectors.toSet())))
-
-        // step.2 wait all replicas to be synced
+                admin
+                    .moveToBrokers(
+                        replicas.stream()
+                            .sorted(Comparator.comparing(Replica::isPreferredLeader).reversed())
+                            .collect(
+                                Collectors.groupingBy(
+                                    ReplicaInfo::topicPartition,
+                                    Collectors.mapping(
+                                        r -> r.nodeInfo().id(), Collectors.toList()))))
+                    .thenApply(ignored -> replicas))
+        // step 2: wait replicas get reassigned
         .thenCompose(
             replicas ->
                 admin
-                    .waitReplicasSynced(replicas, ChronoUnit.DECADES.getDuration())
+                    .waitCluster(
+                        logAllocation.topics(),
+                        clusterInfo ->
+                            clusterInfo
+                                .topicPartitionReplicas()
+                                .containsAll(logAllocation.topicPartitionReplicas()),
+                        timeout,
+                        1)
                     .thenApply(
-                        ignored ->
-                            replicas.stream()
-                                .map(TopicPartitionReplica::topicPartition)
-                                .collect(Collectors.toSet())))
-        // step.3 re-elect leaders
+                        done -> {
+                          if (!done)
+                            throw new IllegalStateException(
+                                "Failed to move "
+                                    + replicas.stream()
+                                        .map(ReplicaInfo::topicPartitionReplica)
+                                        .collect(Collectors.toSet()));
+                          return replicas;
+                        }))
+        // step.3 move replicas to specify folders
+        .thenCompose(
+            replicas ->
+                admin.moveToFolders(
+                    replicas.stream()
+                        .collect(
+                            Collectors.toMap(ReplicaInfo::topicPartitionReplica, Replica::path))))
+        // step.4 wait replicas get synced
+        .thenCompose(
+            replicas -> admin.waitReplicasSynced(logAllocation.topicPartitionReplicas(), timeout))
+        // step.5 re-elect leaders
         .thenCompose(
             topicPartitions ->
                 admin
-                    .preferredLeaderElection(topicPartitions)
+                    .preferredLeaderElection(logAllocation.topicPartitions())
                     .thenCompose(
                         ignored ->
                             admin.waitPreferredLeaderSynced(
-                                topicPartitions, ChronoUnit.DECADES.getDuration()))
-                    .thenAccept(c -> assertion(c, "Failed to sync " + topicPartitions)));
-  }
-
-  public CompletionStage<Void> run(Admin admin, List<Replica> replicas) {
-    var moveBrokerRequest =
-        replicas.stream()
-            .sorted(Comparator.comparing(Replica::isPreferredLeader).reversed())
-            .collect(
-                Collectors.groupingBy(
-                    ReplicaInfo::topicPartition,
-                    Collectors.mapping(r -> r.nodeInfo().id(), Collectors.toList())));
-    var moveFolderRequest =
-        replicas.stream()
-            .collect(Collectors.toMap(ReplicaInfo::topicPartitionReplica, Replica::path));
-    return admin
-        // step.1 move replica to specify brokers
-        .moveToBrokers(moveBrokerRequest)
-        // step.2 wait assignment
-        .thenCompose(
-            ignored ->
-                admin.waitCluster(
-                    moveBrokerRequest.keySet().stream()
-                        .map(TopicPartition::topic)
-                        .collect(Collectors.toSet()),
-                    clusterInfo ->
-                        replicas.stream()
-                            .allMatch(
-                                r ->
-                                    clusterInfo
-                                        .replicaStream()
-                                        .anyMatch(
-                                            replica ->
-                                                replica
-                                                    .topicPartitionReplica()
-                                                    .equals(r.topicPartitionReplica()))),
-                    Duration.ofSeconds(15),
-                    2))
-        .thenAccept(done -> assertion(done, "Failed to sync " + moveFolderRequest.keySet()))
-        // step.3 move replica to specify folder
-        .thenCompose(ignored -> admin.moveToFolders(moveFolderRequest));
+                                logAllocation.topicPartitions(), timeout))
+                    .thenAccept(c -> assertion(c, "Failed to re-election for " + topicPartitions)));
   }
 
-  private void assertion(boolean condition, String info) {
+  static void assertion(boolean condition, String info) {
     if (!condition) throw new IllegalStateException(info);
   }
 }
diff --git a/common/src/main/java/org/astraea/common/consumer/ConsumerConfigs.java b/common/src/main/java/org/astraea/common/consumer/ConsumerConfigs.java
index c1476076..99db07a7 100644
--- a/common/src/main/java/org/astraea/common/consumer/ConsumerConfigs.java
+++ b/common/src/main/java/org/astraea/common/consumer/ConsumerConfigs.java
@@ -17,6 +17,7 @@
 package org.astraea.common.consumer;
 
 import java.util.Set;
+import java.util.TreeSet;
 import org.astraea.common.Utils;
 
 public final class ConsumerConfigs {
@@ -66,7 +67,7 @@ public final class ConsumerConfigs {
   public static final String ALLOW_AUTO_CREATE_TOPICS_CONFIG = "allow.auto.create.topics";
 
   public static final Set<String> ALL_CONFIGS =
-      Utils.constants(ConsumerConfigs.class, name -> name.endsWith("CONFIG"));
+      new TreeSet<>(Utils.constants(ConsumerConfigs.class, name -> name.endsWith("CONFIG")));
 
   // ---------------------------------[Values]---------------------------------//
   public static final String AUTO_OFFSET_RESET_LATEST = "latest";
diff --git a/common/src/main/java/org/astraea/common/producer/ProducerConfigs.java b/common/src/main/java/org/astraea/common/producer/ProducerConfigs.java
index cf2ce24a..655a8110 100644
--- a/common/src/main/java/org/astraea/common/producer/ProducerConfigs.java
+++ b/common/src/main/java/org/astraea/common/producer/ProducerConfigs.java
@@ -17,6 +17,7 @@
 package org.astraea.common.producer;
 
 import java.util.Set;
+import java.util.TreeSet;
 import org.astraea.common.Utils;
 
 public final class ProducerConfigs {
@@ -66,7 +67,7 @@ public final class ProducerConfigs {
   public static final String SECURITY_PROVIDERS_CONFIG = "security.providers";
 
   public static final Set<String> ALL_CONFIGS =
-      Utils.constants(ProducerConfigs.class, name -> name.endsWith("CONFIG"));
+      new TreeSet<>(Utils.constants(ProducerConfigs.class, name -> name.endsWith("CONFIG")));
 
   // ---------------------------------[values]---------------------------------//
   public static final String COMPRESSION_TYPE_NONE = "none";
diff --git a/gui/src/main/java/org/astraea/gui/pane/PaneBuilder.java b/gui/src/main/java/org/astraea/gui/pane/PaneBuilder.java
index 0510ea21..203b55fd 100644
--- a/gui/src/main/java/org/astraea/gui/pane/PaneBuilder.java
+++ b/gui/src/main/java/org/astraea/gui/pane/PaneBuilder.java
@@ -139,9 +139,10 @@ public class PaneBuilder {
                     ? Map.<String, Optional<String>>of()
                     : secondLattice.contents();
             var input = Input.of(List.of(), text);
-            try {
-              checkbox.setSelected(false);
 
+            checkbox.setDisable(true);
+            checkbox.setSelected(false);
+            try {
               var invalidKeys = secondLattice == null ? Set.of() : secondLattice.invalidKeys();
               if (!invalidKeys.isEmpty()) {
                 console.text("please check fields: " + invalidKeys);
@@ -149,8 +150,13 @@ public class PaneBuilder {
               }
               tableViewAction
                   .apply(items, input, console::append)
-                  .whenComplete((data, e) -> console.text(e));
+                  .whenComplete(
+                      (data, e) -> {
+                        checkbox.setDisable(false);
+                        console.text(e);
+                      });
             } catch (Exception e) {
+              checkbox.setDisable(false);
               console.text(e);
             }
           });
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 4b54a36a..f4d53c53 100644
--- a/gui/src/main/java/org/astraea/gui/tab/BalancerNode.java
+++ b/gui/src/main/java/org/astraea/gui/tab/BalancerNode.java
@@ -39,7 +39,7 @@ import org.astraea.common.admin.TopicPartition;
 import org.astraea.common.balancer.Balancer;
 import org.astraea.common.balancer.algorithms.AlgorithmConfig;
 import org.astraea.common.balancer.algorithms.GreedyBalancer;
-import org.astraea.common.balancer.executor.StraightPlanExecutor;
+import org.astraea.common.balancer.executor.RebalancePlanExecutor;
 import org.astraea.common.cost.HasClusterCost;
 import org.astraea.common.cost.MoveCost;
 import org.astraea.common.cost.ReplicaLeaderCost;
@@ -240,14 +240,28 @@ public class BalancerNode {
             plan.proposal().rebalancePlan().replicas().stream()
                 .filter(r -> selectedPartitions.contains(r.topicPartition()))
                 .collect(Collectors.toList());
-        return new StraightPlanExecutor()
-            .run(context.admin(), replicas)
-            .thenAccept(
-                ignored ->
-                    logger.log(
-                        "succeed to balance cluster by moving "
-                            + selectedPartitions.size()
-                            + " partitions"));
+        return context
+            .admin()
+            .addingReplicas(
+                selectedPartitions.stream().map(TopicPartition::topic).collect(Collectors.toSet()))
+            .thenCompose(
+                addingReplicas -> {
+                  if (addingReplicas.isEmpty())
+                    return RebalancePlanExecutor.of()
+                        .run(context.admin(), ClusterInfo.of(replicas), Duration.ofHours(1))
+                        .thenAccept(
+                            ignored ->
+                                logger.log(
+                                    "succeed to balance cluster by moving "
+                                        + selectedPartitions.size()
+                                        + " partitions"));
+                  return CompletableFuture.failedFuture(
+                      new IllegalArgumentException(
+                          "Please wait migrating partitions: "
+                              + addingReplicas.stream()
+                                  .map(r -> r.topic() + "-" + r.partition())
+                                  .collect(Collectors.joining(","))));
+                });
       }
       return CompletableFuture.completedFuture(null);
     };
diff --git a/gui/src/main/java/org/astraea/gui/tab/topic/ReplicaNode.java b/gui/src/main/java/org/astraea/gui/tab/topic/ReplicaNode.java
index 6b4c59e1..72ad08ce 100644
--- a/gui/src/main/java/org/astraea/gui/tab/topic/ReplicaNode.java
+++ b/gui/src/main/java/org/astraea/gui/tab/topic/ReplicaNode.java
@@ -70,6 +70,7 @@ public class ReplicaNode {
               result.put("isPreferredLeader", replica.isPreferredLeader());
               result.put("isOffline", replica.isOffline());
               result.put("isFuture", replica.isFuture());
+              result.put("inSync", replica.inSync());
               result.put("lag", replica.lag());
               result.put("size", DataSize.Byte.of(replica.size()));
               if (leaderSize > replica.size()) {


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 dc0c781b..5d39d3db 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CountDownLatch;
@@ -476,10 +477,16 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var theExecutor =
           new NoOpExecutor() {
             @Override
-            public CompletionStage<Void> run(Admin admin, ClusterLogAllocation targetAllocation) {
-              super.run(admin, targetAllocation);
-              Utils.sleep(Duration.ofSeconds(10));
-              return null;
+            public CompletionStage<Void> run(
+                Admin admin, ClusterInfo<Replica> targetAllocation, Duration timeout) {
+              return super.run(admin, targetAllocation, Duration.ofSeconds(5))
+                  // Use another thread to block this completion to avoid deadlock in
+                  // BalancerHandler#put
+                  .thenApplyAsync(
+                      i -> {
+                        Utils.sleep(Duration.ofSeconds(10));
+                        return i;
+                      });
             }
           };
       var handler = new BalancerHandler(admin, theExecutor);
@@ -602,10 +609,18 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
             final CountDownLatch latch = new CountDownLatch(1);
 
             @Override
-            public CompletionStage<Void> run(Admin admin, ClusterLogAllocation targetAllocation) {
-              super.run(admin, targetAllocation);
-              Utils.packException(() -> latch.await());
-              return null;
+            public CompletionStage<Void> run(
+                Admin admin, ClusterInfo<Replica> targetAllocation, Duration timeout) {
+              return super.run(admin, targetAllocation, Duration.ofSeconds(5))
+                  // Use another thread to block this completion to avoid deadlock in
+                  // BalancerHandler#put
+                  .thenApplyAsync(
+                      i -> {
+                        System.out.println("before block");
+                        Utils.packException(() -> latch.await());
+                        System.out.println("after block");
+                        return i;
+                      });
             }
           };
       var handler = new BalancerHandler(admin, theExecutor);
@@ -647,8 +662,10 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       Assertions.assertNull(progress1.exception);
 
       // it is done
+      System.out.println("before countDown");
       theExecutor.latch.countDown();
-      Utils.sleep(Duration.ofMillis(500));
+      System.out.println("after countDown");
+      Utils.sleep(Duration.ofSeconds(1));
       var progress2 =
           Assertions.assertInstanceOf(
               BalancerHandler.PlanExecutionProgress.class,
@@ -668,9 +685,11 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var theExecutor =
           new NoOpExecutor() {
             @Override
-            public CompletionStage<Void> run(Admin admin, ClusterLogAllocation targetAllocation) {
-              super.run(admin, targetAllocation);
-              throw new RuntimeException("Boom");
+            public CompletionStage<Void> run(
+                Admin admin, ClusterInfo<Replica> targetAllocation, Duration timeout) {
+              return super.run(admin, targetAllocation, Duration.ofSeconds(5))
+                  .thenCompose(
+                      ignored -> CompletableFuture.failedFuture(new RuntimeException("Boom")));
             }
           };
       var handler = new BalancerHandler(admin, theExecutor);
@@ -802,9 +821,10 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
     private final LongAdder executionCounter = new LongAdder();
 
     @Override
-    public CompletionStage<Void> run(Admin admin, ClusterLogAllocation targetAllocation) {
+    public CompletionStage<Void> run(
+        Admin admin, ClusterInfo<Replica> targetAllocation, Duration timeout) {
       executionCounter.increment();
-      return null;
+      return CompletableFuture.completedFuture(null);
     }
 
     int count() {
diff --git a/common/src/test/java/org/astraea/common/admin/BrokerConfigsTest.java b/common/src/test/java/org/astraea/common/admin/BrokerConfigsTest.java
index 8a03c7ff..c4bbc019 100644
--- a/common/src/test/java/org/astraea/common/admin/BrokerConfigsTest.java
+++ b/common/src/test/java/org/astraea/common/admin/BrokerConfigsTest.java
@@ -16,6 +16,7 @@
  */
 package org.astraea.common.admin;
 
+import org.astraea.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -41,6 +42,6 @@ public class BrokerConfigsTest {
   void testDuplicate() {
     Assertions.assertEquals(
         BrokerConfigs.DYNAMICAL_CONFIGS.size(),
-        BrokerConfigs.DYNAMICAL_CONFIGS.stream().distinct().count());
+        Utils.constants(BrokerConfigs.class, name -> name.endsWith("CONFIG")).size());
   }
 }
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 48623fe8..314d431c 100644
--- a/common/src/test/java/org/astraea/common/balancer/BalancerTest.java
+++ b/common/src/test/java/org/astraea/common/balancer/BalancerTest.java
@@ -102,7 +102,7 @@ class BalancerTest extends RequireBrokerCluster {
                   admin.brokerFolders().toCompletableFuture().join())
               .orElseThrow();
       new StraightPlanExecutor()
-          .run(admin, plan.proposal().rebalancePlan())
+          .run(admin, plan.proposal().rebalancePlan(), Duration.ofSeconds(10))
           .toCompletableFuture()
           .join();
 
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 71a05293..8af4545b 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
@@ -16,7 +16,7 @@
  */
 package org.astraea.common.balancer.algorithms;
 
-import java.util.Set;
+import org.astraea.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -35,8 +35,8 @@ class GreedyBalancerTest {
         "Config exists for backward compatability reason");
 
     Assertions.assertEquals(
-        Set.copyOf(GreedyBalancer.ALL_CONFIGS).size(),
         GreedyBalancer.ALL_CONFIGS.size(),
+        Utils.constants(GreedyBalancer.class, name -> name.endsWith("CONFIG")).size(),
         "No duplicate element");
   }
 }
diff --git a/common/src/test/java/org/astraea/common/balancer/algorithms/SingleStepBalancerTest.java b/common/src/test/java/org/astraea/common/balancer/algorithms/SingleStepBalancerTest.java
index 98450a4a..c5be4bef 100644
--- a/common/src/test/java/org/astraea/common/balancer/algorithms/SingleStepBalancerTest.java
+++ b/common/src/test/java/org/astraea/common/balancer/algorithms/SingleStepBalancerTest.java
@@ -16,7 +16,7 @@
  */
 package org.astraea.common.balancer.algorithms;
 
-import java.util.Set;
+import org.astraea.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -35,8 +35,8 @@ class SingleStepBalancerTest {
         "Config exists for backward compatability reason");
 
     Assertions.assertEquals(
-        Set.copyOf(SingleStepBalancer.ALL_CONFIGS).size(),
         SingleStepBalancer.ALL_CONFIGS.size(),
+        Utils.constants(SingleStepBalancer.class, name -> name.endsWith("CONFIG")).size(),
         "No duplicate element");
   }
 }
diff --git a/common/src/test/java/org/astraea/common/balancer/executor/StraightPlanExecutorTest.java b/common/src/test/java/org/astraea/common/balancer/executor/StraightPlanExecutorTest.java
index bf0ca630..5a48c965 100644
--- a/common/src/test/java/org/astraea/common/balancer/executor/StraightPlanExecutorTest.java
+++ b/common/src/test/java/org/astraea/common/balancer/executor/StraightPlanExecutorTest.java
@@ -104,7 +104,8 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
       final var expectedAllocation = ClusterLogAllocation.of(ClusterInfo.of(allocation));
       final var expectedTopicPartition = expectedAllocation.topicPartitions();
 
-      var execute = new StraightPlanExecutor().run(admin, expectedAllocation);
+      var execute =
+          new StraightPlanExecutor().run(admin, expectedAllocation, Duration.ofSeconds(10));
 
       execute.toCompletableFuture().join();
 
diff --git a/gui/src/test/java/org/astraea/gui/tab/topic/ReplicaTabTest.java b/gui/src/test/java/org/astraea/gui/tab/topic/ReplicaNodeTest.java
similarity index 99%
rename from gui/src/test/java/org/astraea/gui/tab/topic/ReplicaTabTest.java
rename to gui/src/test/java/org/astraea/gui/tab/topic/ReplicaNodeTest.java
index 713d4d98..69d06858 100644
--- a/gui/src/test/java/org/astraea/gui/tab/topic/ReplicaTabTest.java
+++ b/gui/src/test/java/org/astraea/gui/tab/topic/ReplicaNodeTest.java
@@ -33,7 +33,7 @@ import org.astraea.it.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class ReplicaTabTest extends RequireBrokerCluster {
+public class ReplicaNodeTest extends RequireBrokerCluster {
 
   @Test
   void testTableAction() {