Repository: skiptests/astraea
Pull request #62 (2021-10-20 13:59:47)

Rewrite ReplicaCollieTest by embedded server
- 重寫`ReplicaCollieTest`
- 穩定`TopicAdminTest`


Patch:
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
index 741e3e25..94e003db 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -6,6 +6,9 @@ import java.io.UncheckedIOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.file.Files;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 public final class Utils {
@@ -63,5 +66,33 @@ public final class Utils {
     }
   }
 
+  /**
+   * Wait for procedure. Default is 10 seconds
+   *
+   * @param done a flag indicating the result.
+   */
+  public static void waitFor(Supplier<Boolean> done) {
+    waitFor(done, Duration.ofSeconds(10));
+  }
+
+  /**
+   * Wait for procedure.
+   *
+   * @param done a flag indicating the result.
+   */
+  public static void waitFor(Supplier<Boolean> done, Duration timeout) {
+    var endTime = System.currentTimeMillis() + timeout.toMillis();
+    Exception lastError = null;
+    while (System.currentTimeMillis() <= endTime)
+      try {
+        if (done.get()) return;
+        TimeUnit.SECONDS.sleep(1);
+      } catch (Exception e) {
+        lastError = e;
+      }
+    if (lastError != null) throw new RuntimeException(lastError);
+    throw new RuntimeException("Timeout to wait procedure");
+  }
+
   private Utils() {}
 }
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index b1fe5124..771049e4 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -3,6 +3,7 @@ package org.astraea.topic;
 import java.io.Closeable;
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.Node;
@@ -20,31 +21,51 @@ public interface TopicAdmin extends Closeable {
     return new TopicAdmin() {
 
       @Override
-      public void createTopic(String topic, int numberOfPartitions) {
+      public void createTopic(String topic, int numberOfPartitions, short numberOfReplicas) {
         var topics = topics();
         if (topics.contains(topic)) {
           var partitions = partitions(Set.of(topic));
+          if (partitions.size() > numberOfPartitions)
+            throw new IllegalArgumentException(
+                "Reducing the number of partitions is disallowed. Current: "
+                    + partitions.size()
+                    + " requested: "
+                    + numberOfPartitions);
+
+          var allBrokers = brokerIds();
+          if (allBrokers.size() < numberOfReplicas)
+            throw new IllegalArgumentException(
+                "expected number of replicas is "
+                    + numberOfReplicas
+                    + ", but there are only "
+                    + allBrokers.size()
+                    + " brokers");
+
           if (partitions.size() < numberOfPartitions) {
             Utils.handleException(
                 () ->
                     admin
                         .createPartitions(
-                            Map.of(topic, NewPartitions.increaseTo(numberOfPartitions)))
+                            Map.of(
+                                topic,
+                                NewPartitions.increaseTo(
+                                    numberOfPartitions,
+                                    IntStream.range(0, numberOfPartitions - partitions.size())
+                                        .mapToObj(
+                                            i ->
+                                                new ArrayList<>(allBrokers)
+                                                    .subList(0, numberOfReplicas))
+                                        .collect(Collectors.toList()))))
                         .all()
                         .get());
           }
-          if (partitions.size() > numberOfPartitions) {
-            throw new IllegalArgumentException(
-                "Reducing the number of partitions is disallowed. Current: "
-                    + partitions.size()
-                    + " requested: "
-                    + numberOfPartitions);
-          }
+
         } else {
           Utils.handleException(
               () ->
                   admin
-                      .createTopics(List.of(new NewTopic(topic, numberOfPartitions, (short) 1)))
+                      .createTopics(
+                          List.of(new NewTopic(topic, numberOfPartitions, numberOfReplicas)))
                       .all()
                       .get());
         }
@@ -204,11 +225,11 @@ public interface TopicAdmin extends Closeable {
 
       @Override
       public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
-        var replicaInfo =
+        var replicaInfos =
             Utils.handleException(() -> admin.describeLogDirs(brokerIds()).allDescriptions().get());
 
         var replicaLags =
-            replicaInfo.entrySet().stream()
+            replicaInfos.entrySet().stream()
                 .collect(
                     Collectors.toMap(
                         Map.Entry::getKey,
@@ -221,7 +242,7 @@ public interface TopicAdmin extends Closeable {
                                     Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))));
 
         var replicaPaths =
-            replicaInfo.entrySet().stream()
+            replicaInfos.entrySet().stream()
                 .collect(
                     Collectors.toMap(
                         Map.Entry::getKey,
@@ -255,30 +276,33 @@ public interface TopicAdmin extends Closeable {
                                                 e.getKey(), topicPartitionInfo.partition()),
                                             topicPartitionInfo.replicas().stream()
                                                 .map(
-                                                    node ->
-                                                        new Replica(
-                                                            node.id(),
-                                                            Optional.ofNullable(
-                                                                    replicaLags
-                                                                        .getOrDefault(
-                                                                            node.id(), Map.of())
-                                                                        .get(
-                                                                            new TopicPartition(
-                                                                                e.getKey(),
-                                                                                topicPartitionInfo
-                                                                                    .partition())))
-                                                                .map(ReplicaInfo::offsetLag)
-                                                                .orElse(-1L),
-                                                            topicPartitionInfo.leader().id()
-                                                                == node.id(),
-                                                            topicPartitionInfo.isr().contains(node),
-                                                            replicaPaths
-                                                                .get(node.id())
-                                                                .get(
-                                                                    new TopicPartition(
-                                                                        e.getKey(),
-                                                                        topicPartitionInfo
-                                                                            .partition()))))
+                                                    node -> {
+                                                      var replicaInfo =
+                                                          replicaLags
+                                                              .getOrDefault(node.id(), Map.of())
+                                                              .get(
+                                                                  new TopicPartition(
+                                                                      e.getKey(),
+                                                                      topicPartitionInfo
+                                                                          .partition()));
+                                                      return new Replica(
+                                                          node.id(),
+                                                          replicaInfo == null
+                                                              ? -1
+                                                              : replicaInfo.offsetLag(),
+                                                          topicPartitionInfo.leader().id()
+                                                              == node.id(),
+                                                          topicPartitionInfo.isr().contains(node),
+                                                          replicaInfo != null
+                                                              && replicaInfo.isFuture(),
+                                                          replicaPaths
+                                                              .get(node.id())
+                                                              .get(
+                                                                  new TopicPartition(
+                                                                      e.getKey(),
+                                                                      topicPartitionInfo
+                                                                          .partition())));
+                                                    })
                                                 .sorted(
                                                     Comparator.comparing((Replica r) -> r.broker))
                                                 .collect(Collectors.toList()))))
@@ -298,7 +322,20 @@ public interface TopicAdmin extends Closeable {
    * @param topic topic name
    * @param numberOfPartitions expected number of partitions.
    */
-  void createTopic(String topic, int numberOfPartitions);
+  default void createTopic(String topic, int numberOfPartitions) {
+    createTopic(topic, numberOfPartitions, (short) 1);
+  }
+
+  /**
+   * make sure there is a topic having requested name and requested number of partitions. If the
+   * topic is existent and the number of partitions is larger than requested number, it will throw
+   * exception.
+   *
+   * @param topic topic name
+   * @param numberOfPartitions expected number of partitions.
+   * @param numberOfReplicas expected number of replicas.
+   */
+  void createTopic(String topic, int numberOfPartitions, short numberOfReplicas);
 
   /**
    * @param topics topic names
@@ -322,6 +359,8 @@ public interface TopicAdmin extends Closeable {
   Set<Integer> brokerIds();
 
   /**
+   * Assign the topic partition to specific brokers.
+   *
    * @param topicName topic name
    * @param partition partition
    * @param brokers to hold all the
@@ -404,13 +443,16 @@ public interface TopicAdmin extends Closeable {
     public final long lag;
     public final boolean leader;
     public final boolean inSync;
+    public final boolean isFuture;
     public final String path;
 
-    public Replica(int broker, long lag, boolean leader, boolean inSync, String path) {
+    public Replica(
+        int broker, long lag, boolean leader, boolean inSync, boolean isFuture, String path) {
       this.broker = broker;
       this.lag = lag;
       this.leader = leader;
       this.inSync = inSync;
+      this.isFuture = isFuture;
       this.path = path;
     }
 
@@ -425,10 +467,30 @@ public interface TopicAdmin extends Closeable {
           + leader
           + ", inSync="
           + inSync
+          + ", isFuture="
+          + isFuture
           + ", path='"
           + path
           + '\''
           + '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      Replica replica = (Replica) o;
+      return broker == replica.broker
+          && lag == replica.lag
+          && leader == replica.leader
+          && inSync == replica.inSync
+          && isFuture == replica.isFuture
+          && Objects.equals(path, replica.path);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(broker, lag, leader, inSync, isFuture, path);
+    }
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
index 08f92102..96460fee 100644
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
@@ -5,56 +5,29 @@ import static org.junit.jupiter.api.Assertions.*;
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.management.*;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
 import org.astraea.metrics.jmx.MBeanClient;
-import org.astraea.metrics.jmx.Utility;
 import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
 import org.astraea.metrics.kafka.metrics.TotalTimeMs;
+import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
-class KafkaMetricsTest {
+class KafkaMetricsTest extends RequireBrokerCluster {
 
   private MBeanServer mBeanServer;
   private JMXConnectorServer jmxServer;
-  private Map<ObjectName, Object> registeredBeans = new HashMap<>();
   private MBeanClient mBeanClient;
 
-  private void register(ObjectName name, Object mBean) {
-    registeredBeans.put(name, mBean);
-    try {
-      mBeanServer.registerMBean(mBean, name);
-    } catch (InstanceAlreadyExistsException
-        | MBeanRegistrationException
-        | NotCompliantMBeanException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  private void clearRegisteredMBeans() {
-    registeredBeans.forEach(
-        (name, mbeans) -> {
-          try {
-            mBeanServer.unregisterMBean(name);
-          } catch (InstanceNotFoundException | MBeanRegistrationException e) {
-            throw new RuntimeException(e);
-          }
-        });
-    registeredBeans.clear();
-  }
-
   @BeforeEach
   void setUp() throws IOException {
     JMXServiceURL serviceURL = new JMXServiceURL("service:jmx:rmi://127.0.0.1");
@@ -70,7 +43,6 @@ class KafkaMetricsTest {
   @AfterEach
   void tearDown() throws Exception {
     jmxServer.stop();
-    clearRegisteredMBeans();
     mBeanServer = null;
     mBeanClient.close();
   }
@@ -89,149 +61,69 @@ class KafkaMetricsTest {
 
   @ParameterizedTest
   @EnumSource(value = KafkaMetrics.BrokerTopic.class)
-  void testRequestBrokerTopicMetrics(KafkaMetrics.BrokerTopic metric)
-      throws MalformedObjectNameException {
-    // arrange
-    Object mbean =
-        Utility.createReadOnlyDynamicMBean(
-            Map.of(
-                "Count", 1L,
-                "EventType", "bytes",
-                "FifteenMinuteRate", 2.0,
-                "FiveMinuteRate", 4.0,
-                "MeanRate", 8.0,
-                "OneMinuteRate", 16.0,
-                "RateUnit", TimeUnit.SECONDS));
-    register(
-        ObjectName.getInstance("kafka.server:type=BrokerTopicMetrics,name=" + metric.name()),
-        mbean);
-
+  void testRequestBrokerTopicMetrics(KafkaMetrics.BrokerTopic metric) {
     // act
     BrokerTopicMetricsResult result = metric.fetch(mBeanClient);
 
-    // assert
-    assertEquals(1L, result.count());
-    assertEquals("bytes", result.eventType());
-    assertEquals(2.0, result.fifteenMinuteRate());
-    assertEquals(4.0, result.fiveMinuteRate());
-    assertEquals(8.0, result.meanRate());
-    assertEquals(16.0, result.oneMinuteRate());
-    assertEquals(TimeUnit.SECONDS, result.rateUnit());
+    // assert access attribute will not throw casting error
+    // assert attribute actually exists
+    assertDoesNotThrow(result::count);
+    assertDoesNotThrow(result::eventType);
+    assertDoesNotThrow(result::fifteenMinuteRate);
+    assertDoesNotThrow(result::fiveMinuteRate);
+    assertDoesNotThrow(result::meanRate);
+    assertDoesNotThrow(result::oneMinuteRate);
+    assertDoesNotThrow(result::rateUnit);
   }
 
   @ParameterizedTest()
   @EnumSource(value = KafkaMetrics.Purgatory.class)
-  void testPurgatorySize(KafkaMetrics.Purgatory request) throws MalformedObjectNameException {
-    // arrange
-    Function<Integer, Object> fMbean =
-        (a) -> Utility.createReadOnlyDynamicMBean(Map.of("Value", a));
-    Function<String, String> fname =
-        (a) ->
-            String.format(
-                "kafka.server:type=DelayedOperationPurgatory,delayedOperation=%s,name=PurgatorySize",
-                a);
-    register(ObjectName.getInstance(fname.apply(request.name())), fMbean.apply(0xcafebabe));
-
-    // act
-    int size = request.size(mBeanClient);
-
-    // assert
-    assertEquals(0xcafebabe, size);
+  void testPurgatorySize(KafkaMetrics.Purgatory request) {
+    // act assert type casting correct and field exists
+    assertDoesNotThrow(() -> request.size(mBeanClient));
   }
 
   @ParameterizedTest()
   @EnumSource(value = KafkaMetrics.Request.class)
-  void testRequestTotalTimeMs(KafkaMetrics.Request request) throws MalformedObjectNameException {
-    // arrange
-    Map<String, Object> map = new HashMap<>();
-    map.put("50thPercentile", 1.0);
-    map.put("75thPercentile", 2.0);
-    map.put("95thPercentile", 3.0);
-    map.put("98thPercentile", 4.0);
-    map.put("99thPercentile", 5.0);
-    map.put("999thPercentile", 6.0);
-    map.put("Count", 7L);
-    map.put("Max", 8.0);
-    map.put("Mean", 9.0);
-    map.put("Min", 10.0);
-    map.put("StdDev", 11.0);
-    Object mbean = Utility.createReadOnlyDynamicMBean(map);
-    register(
-        ObjectName.getInstance(
-            "kafka.network:type=RequestMetrics,request=" + request.name() + ",name=TotalTimeMs"),
-        mbean);
-
+  void testRequestTotalTimeMs(KafkaMetrics.Request request) {
     // act
     TotalTimeMs totalTimeMs = request.totalTimeMs(mBeanClient);
 
-    // assert
-    assertEquals(totalTimeMs.percentile50(), 1.0);
-    assertEquals(totalTimeMs.percentile75(), 2.0);
-    assertEquals(totalTimeMs.percentile95(), 3.0);
-    assertEquals(totalTimeMs.percentile98(), 4.0);
-    assertEquals(totalTimeMs.percentile99(), 5.0);
-    assertEquals(totalTimeMs.percentile999(), 6.0);
-    assertEquals(totalTimeMs.count(), 7L);
-    assertEquals(totalTimeMs.max(), 8.0);
-    assertEquals(totalTimeMs.mean(), 9.0);
-    assertEquals(totalTimeMs.min(), 10.0);
-    assertEquals(totalTimeMs.stdDev(), 11.0);
+    // assert type casting correct and field exists
+    assertDoesNotThrow(totalTimeMs::percentile50);
+    assertDoesNotThrow(totalTimeMs::percentile75);
+    assertDoesNotThrow(totalTimeMs::percentile95);
+    assertDoesNotThrow(totalTimeMs::percentile98);
+    assertDoesNotThrow(totalTimeMs::percentile99);
+    assertDoesNotThrow(totalTimeMs::percentile999);
+    assertDoesNotThrow(totalTimeMs::count);
+    assertDoesNotThrow(totalTimeMs::max);
+    assertDoesNotThrow(totalTimeMs::mean);
+    assertDoesNotThrow(totalTimeMs::min);
+    assertDoesNotThrow(totalTimeMs::stdDev);
   }
 
   @Test
-  void testGlobalPartitionCount() throws MalformedObjectNameException {
-    // arrange
-    Object mbean = Utility.createReadOnlyDynamicMBean(Map.of("Value", 0xcafebabe));
-    register(
-        ObjectName.getInstance("kafka.controller:type=KafkaController,name=GlobalPartitionCount"),
-        mbean);
-
+  void testGlobalPartitionCount() {
     // act
-    int integer = KafkaMetrics.TopicPartition.globalPartitionCount(mBeanClient);
-
-    // assert
-    assertEquals(0xcafebabe, integer);
+    assertDoesNotThrow(() -> KafkaMetrics.TopicPartition.globalPartitionCount(mBeanClient));
   }
 
   @Test
-  void testUnderReplicatedPartitions() throws MalformedObjectNameException {
-    // arrange
-    Object mbean = Utility.createReadOnlyDynamicMBean(Map.of("Value", 0x55665566));
-    register(
-        ObjectName.getInstance("kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"),
-        mbean);
-
-    // act
-    int integer = KafkaMetrics.TopicPartition.underReplicatedPartitions(mBeanClient);
-
-    // assert
-    assertEquals(0x55665566, integer);
+  void testUnderReplicatedPartitions() {
+    assertDoesNotThrow(() -> KafkaMetrics.TopicPartition.underReplicatedPartitions(mBeanClient));
   }
 
   @Test
-  void testSize() throws MalformedObjectNameException {
+  void testSize() throws IOException {
     // arrange
-    String topicName = "A";
-    Function<Long, Object> fMbean =
-        (Long l) -> Utility.createReadOnlyDynamicMBean(Map.of("Value", l));
-    Function<Integer, String> fName =
-        (Integer i) -> "kafka.log:type=Log,name=Size,topic=" + topicName + ",partition=" + i;
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      String topicName = getClass().getName();
+      admin.createTopic(topicName, 10);
 
-    register(ObjectName.getInstance(fName.apply(1)), fMbean.apply(100L));
-    register(ObjectName.getInstance(fName.apply(2)), fMbean.apply(200L));
-    register(ObjectName.getInstance(fName.apply(4)), fMbean.apply(400L));
-    register(ObjectName.getInstance(fName.apply(5)), fMbean.apply(500L));
-
-    // act
-    Map<Integer, Long> size = KafkaMetrics.TopicPartition.size(mBeanClient, topicName);
-
-    // assert
-    assertFalse(size.containsKey(0));
-    assertEquals(100L, size.get(1));
-    assertEquals(200L, size.get(2));
-    assertFalse(size.containsKey(3));
-    assertEquals(400L, size.get(4));
-    assertEquals(500L, size.get(5));
+      // act assert
+      assertDoesNotThrow(() -> KafkaMetrics.TopicPartition.size(mBeanClient, topicName));
+    }
   }
 
   @Test
diff --git a/app/src/test/java/org/astraea/topic/FakeTopicAdmin.java b/app/src/test/java/org/astraea/topic/FakeTopicAdmin.java
index 75b3b92c..61396373 100644
--- a/app/src/test/java/org/astraea/topic/FakeTopicAdmin.java
+++ b/app/src/test/java/org/astraea/topic/FakeTopicAdmin.java
@@ -12,7 +12,7 @@ public abstract class FakeTopicAdmin implements TopicAdmin {
   }
 
   @Override
-  public void createTopic(String topic, int numberOfPartitions) {
+  public void createTopic(String topic, int numberOfPartitions, short numberOfReplicas) {
     throw new UnsupportedOperationException();
   }
 
diff --git a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
index 2650ea2a..757c458c 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
@@ -1,75 +1,70 @@
 package org.astraea.topic;
 
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
+import java.io.IOException;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 import org.apache.kafka.common.TopicPartition;
+import org.astraea.Utils;
+import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class ReplicaCollieTest {
-
-  private final String topicName = "ReplicaCollieTest";
-  private final int partition = 0;
-  private final int badBroker = 10;
-  private final int goodBroker = 100;
-  private final int earliestOffset = 199;
-  private final int latestOffset = 299;
-  private final Set<Integer> reassignment = new HashSet<>();
-  private final TopicAdmin admin =
-      new FakeTopicAdmin() {
-        @Override
-        public Set<String> topics() {
-          return Set.of(topicName);
-        }
-
-        @Override
-        public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
-          return Map.of(
-              new TopicPartition(topicName, partition),
-              List.of(new Replica(badBroker, 0, true, true, "path")));
-        }
-
-        @Override
-        public Set<Integer> brokerIds() {
-          return Set.of(goodBroker, badBroker);
-        }
-
-        @Override
-        public void reassign(String topicName, int partition, Set<Integer> brokers) {
-          reassignment.addAll(brokers);
-        }
-      };
+public class ReplicaCollieTest extends RequireBrokerCluster {
 
   @Test
-  void testVerify() {
+  void testVerify() throws IOException, InterruptedException {
     test(true);
   }
 
   @Test
-  void testExecute() {
+  void testExecute() throws IOException, InterruptedException {
     test(false);
   }
 
-  private void test(boolean verify) {
-    var argument = new ReplicaCollie.Argument();
-    argument.fromBrokers = Set.of(badBroker);
-    argument.toBrokers = Set.of();
-    argument.brokers = "just unit test";
-    argument.verify = verify;
-    var result = ReplicaCollie.execute(admin, argument);
-    Assertions.assertEquals(1, result.size());
-    var assignment = result.get(new TopicPartition(topicName, partition));
-    Assertions.assertEquals(1, assignment.getKey().size());
-    Assertions.assertEquals(badBroker, assignment.getKey().iterator().next());
-    Assertions.assertEquals(1, assignment.getValue().size());
-    Assertions.assertEquals(goodBroker, assignment.getValue().iterator().next());
-    if (verify) {
-      Assertions.assertEquals(0, reassignment.size());
-    } else {
-      Assertions.assertEquals(1, reassignment.size());
-      Assertions.assertEquals(goodBroker, reassignment.iterator().next());
+  private void test(boolean verify) throws IOException, InterruptedException {
+    var topicName = "ReplicaCollieTest";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin.createTopic(topicName, 1, (short) 1);
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(5);
+      var partitionReplicas = topicAdmin.replicas(Set.of(topicName));
+      Assertions.assertEquals(1, partitionReplicas.size());
+      var replicas =
+          partitionReplicas.get(new TopicPartition(topicName, 0)).stream()
+              .filter(r -> !r.isFuture)
+              .collect(Collectors.toList());
+      Assertions.assertEquals(1, replicas.size());
+      var badBroker = replicas.get(0).broker;
+      var argument = new ReplicaCollie.Argument();
+      argument.fromBrokers = Set.of(badBroker);
+      argument.toBrokers = Set.of();
+      argument.brokers = bootstrapServers();
+      argument.verify = verify;
+      var result = ReplicaCollie.execute(topicAdmin, argument);
+      Assertions.assertEquals(1, result.size());
+      var assignment = result.get(new TopicPartition(topicName, 0));
+      Assertions.assertEquals(1, assignment.getKey().size());
+      Assertions.assertEquals(badBroker, assignment.getKey().iterator().next());
+      Assertions.assertEquals(1, assignment.getValue().size());
+      Assertions.assertFalse(assignment.getValue().contains(badBroker));
+      if (verify) {
+        var currentReplicas = topicAdmin.replicas(Set.of(topicName));
+        Assertions.assertEquals(partitionReplicas.size(), currentReplicas.size());
+        partitionReplicas.forEach(
+            (tp, rs) -> {
+              var currentRs = currentReplicas.get(tp);
+              Assertions.assertEquals(rs.size(), currentRs.size());
+              for (var index = 0; index != rs.size(); ++index)
+                Assertions.assertEquals(rs.get(index), currentRs.get(index));
+            });
+      } else {
+        Utils.waitFor(
+            () -> {
+              var rs = topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0));
+              return rs.size() == 1 && rs.stream().noneMatch(r -> r.broker == badBroker);
+            });
+      }
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index 4a9dee80..d76e89a2 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -3,7 +3,9 @@ package org.astraea.topic;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import org.astraea.Utils;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -11,10 +13,12 @@ import org.junit.jupiter.api.Test;
 public class TopicAdminTest extends RequireBrokerCluster {
 
   @Test
-  void testPartitions() throws IOException {
+  void testPartitions() throws IOException, InterruptedException {
     var topicName = "testPartitions";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.createTopic(topicName, 3);
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
       Assertions.assertTrue(topicAdmin.topics().contains(topicName));
       var partitions = topicAdmin.replicas(Set.of(topicName));
       Assertions.assertEquals(3, partitions.size());
@@ -32,10 +36,12 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testGroups() throws IOException {
+  void testGroups() throws IOException, InterruptedException {
     var topicName = "testGroups";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.createTopic(topicName, 3);
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
       Assertions.assertTrue(topicAdmin.topics().contains(topicName));
       var groups = topicAdmin.groups(Set.of(topicName));
       Assertions.assertEquals(0, groups.size());
@@ -43,10 +49,12 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testOffsets() throws IOException {
+  void testOffsets() throws IOException, InterruptedException {
     var topicName = "testOffsets";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
       topicAdmin.createTopic(topicName, 3);
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
       var offsets = topicAdmin.offsets(Set.of(topicName));
       Assertions.assertEquals(3, offsets.size());
       offsets
@@ -58,4 +66,24 @@ public class TopicAdminTest extends RequireBrokerCluster {
               });
     }
   }
+
+  @Test
+  void testReassign() throws IOException, InterruptedException {
+    var topicName = "testReassign";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin.createTopic(topicName, 1);
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
+      var broker = topicAdmin.brokerIds().iterator().next();
+      topicAdmin.reassign(topicName, 0, Set.of(broker));
+      Utils.waitFor(
+          () -> {
+            var replicas = topicAdmin.replicas(Set.of(topicName));
+            var partitionReplicas = replicas.entrySet().iterator().next().getValue();
+            return replicas.size() == 1
+                && partitionReplicas.size() == 1
+                && partitionReplicas.get(0).broker == broker;
+          });
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/topic/TopicExplorerTest.java b/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
index 86632e0c..1ce84087 100644
--- a/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
@@ -39,7 +39,7 @@ public class TopicExplorerTest {
           @Override
           public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
             return Map.of(
-                topicPartition, List.of(new Replica(brokerId, lag, leader, inSync, "path")));
+                topicPartition, List.of(new Replica(brokerId, lag, leader, inSync, false, "path")));
           }
         }) {
       var result = TopicExplorer.execute(admin, Set.of(topicName));