Repository: skiptests/astraea
Pull request #309 (2022-04-23 12:41:10)

add ReplicaManager to KafkaMetrics
 在KafkaMetrics新增一個enum，方便未來cost function取得算分所需得metrics


Patch:
diff --git a/README.md b/README.md
index b1913727..e71a5914 100644
--- a/README.md
+++ b/README.md
@@ -342,25 +342,25 @@ This tool offers an effective way to migrate specify replicas from specific brok
 ### Move all replicas of topic "abc" from broker_0 to broker_1
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0,1 --to 2,3 --topics abc"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topics abc"
 ```
 
 ### Move all replicas of topic "abc" in broker_0 to other folders
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 0 -topics abc"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 0 --topics abc"
 ```
 
 ### Move specify replicas of topic "abc" and "def" from broker_0 to broker_1
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc --partitions 0,1"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc,def --partitions 0,1"
 ```
 
 ### Move specify replicas of topic "abc"  and "def" from broker_0 to broker_1 specify folder
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc --partitions 0,1 --path /tmp/log1"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc,def --partitions 0,1 --path /tmp/log1"
 ```
 
 ### Replica Collie Configurations
diff --git a/app/src/main/java/org/astraea/metrics/kafka/HasValue.java b/app/src/main/java/org/astraea/metrics/kafka/HasValue.java
new file mode 100644
index 00000000..6f0203d4
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/HasValue.java
@@ -0,0 +1,15 @@
+package org.astraea.metrics.kafka;
+
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.BeanObject;
+
+public interface HasValue extends HasBeanObject {
+  default long value() {
+    var value = beanObject().getAttributes().getOrDefault("Value", 0);
+    return ((Number) value).longValue();
+  }
+
+  static HasValue of(BeanObject beanObject) {
+    return () -> beanObject;
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
index 0b1e2a47..19587e5c 100644
--- a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
@@ -4,6 +4,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.java.JvmMemory;
 import org.astraea.metrics.java.OperatingSystemInfo;
 import org.astraea.metrics.jmx.BeanObject;
@@ -120,13 +121,36 @@ public final class KafkaMetrics {
   }
 
   public enum Purgatory {
-    AlterAcls,
-    DeleteRecords,
-    ElectLeader,
-    Fetch,
-    Heartbeat,
-    Produce,
-    Rebalance;
+    AlterAcls("AlterAcls"),
+    DeleteRecords("DeleteRecords"),
+    ElectLeader("ElectLeader"),
+    Fetch("Fetch"),
+    Heartbeat("Heartbeat"),
+    Produce("Produce"),
+    Rebalance("Rebalance");
+
+    private final String metricName;
+
+    Purgatory(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder("kafka.server")
+                  .property("type", "DelayedOperationPurgatory")
+                  .property("delayedOperation", metricName)
+                  .property("name", "PurgatorySize")
+                  .build())
+          .stream()
+          .map(HasValue::of)
+          .collect(Collectors.toUnmodifiableList());
+    }
 
     public int size(MBeanClient mBeanClient) {
       return (int)
@@ -158,9 +182,79 @@ public final class KafkaMetrics {
     }
   }
 
-  public static final class TopicPartition {
+  public enum ReplicaManager {
+    AtMinIsrPartitionCount("AtMinIsrPartitionCount"),
+    LeaderCount("LeaderCount"),
+    OfflineReplicaCount("OfflineReplicaCount"),
+    PartitionCount("PartitionCount"),
+    ReassigningPartitions("ReassigningPartitions"),
+    UnderMinIsrPartitionCount("UnderMinIsrPartitionCount"),
+    UnderReplicatedPartitions("UnderReplicatedPartition");
+    private final String metricName;
+
+    ReplicaManager(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
 
-    private TopicPartition() {}
+    public static ReplicaManager of(String metricName) {
+      return Arrays.stream(ReplicaManager.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder("kafka.server")
+                  .property("type", "ReplicaManager")
+                  .property("name", metricName)
+                  .build())
+          .stream()
+          .map(HasValue::of)
+          .collect(Collectors.toUnmodifiableList());
+    }
+  }
+
+  public enum TopicPartition {
+    LodEndOffset("LodEndOffset"),
+    LogStartOffset("LogStartOffset"),
+    NumLogSegments("NumLogSegments"),
+    Size("Size");
+    private final String metricName;
+
+    TopicPartition(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public static TopicPartition of(String metricName) {
+      return Arrays.stream(TopicPartition.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder("kafka.log")
+                  .property("type", "Log")
+                  .property("topic", "*")
+                  .property("partition", "*")
+                  .property("name", metricName)
+                  .build())
+          .stream()
+          .map(HasValue::of)
+          .collect(Collectors.toUnmodifiableList());
+    }
 
     /**
      * Number of partitions across all topics in the cluster.
@@ -207,17 +301,15 @@ public final class KafkaMetrics {
      *     pair of partition id and its log size
      */
     public static Map<Integer, Long> size(MBeanClient client, String topicName) {
-      Collection<BeanObject> beanObjects =
-          client.queryBeans(
+      return client
+          .queryBeans(
               BeanQuery.builder("kafka.log")
                   .property("type", "Log")
                   .property("topic", topicName)
                   .property("partition", "*")
                   .property("name", "Size")
-                  .build());
-
-      // collect result as a map.
-      return beanObjects.stream()
+                  .build())
+          .stream()
           .collect(
               Collectors.toMap(
                   (BeanObject a) -> Integer.parseInt(a.getProperties().get("partition")),
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index 3d63ba4f..e89c7e29 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -12,6 +12,9 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -19,12 +22,14 @@ import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.clients.admin.ElectLeadersResult;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
 import org.apache.kafka.clients.admin.MemberDescription;
 import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.admin.ReplicaInfo;
+import org.apache.kafka.common.ElectionType;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.TopicPartitionReplica;
@@ -73,7 +78,7 @@ public class Builder {
     }
 
     @Override
-    public Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers) {
+    public Map<Integer, Set<String>> brokerFolders(Collection<Integer> brokers) {
       return Utils.handleException(
           () ->
               admin.describeLogDirs(brokers).allDescriptions().get().entrySet().stream()
@@ -231,6 +236,33 @@ public class Builder {
                   .collect(Collectors.toSet()));
     }
 
+    @Override
+    public Map<TopicPartition, Boolean> changeReplicaLeader(
+        Map<TopicPartition, Integer> partitions) {
+      Map<TopicPartition, Boolean> result = new HashMap<>();
+      partitions.forEach(
+          (tp, broker) -> {
+            var brokers =
+                replicas(Set.of(tp.topic())).get(tp).stream()
+                    .map(Replica::broker)
+                    .collect(Collectors.toList());
+            if (!brokers.contains(broker))
+              throw new IllegalArgumentException("replica " + tp + " is not in broker " + broker);
+            brokers.remove(broker);
+            brokers.add(0, broker);
+            migrator().partition(tp.topic(), tp.partition()).moveTo(brokers);
+            ElectLeadersResult electLeadersResult =
+                admin.electLeaders(ElectionType.PREFERRED, Set.of(tp));
+            try {
+              electLeadersResult.all().get(10L, TimeUnit.SECONDS);
+            } catch (InterruptedException | TimeoutException | ExecutionException e) {
+              throw new RuntimeException(e);
+            }
+            result.put(tp, electLeadersResult.partitions().isDone());
+          });
+      return result;
+    }
+
     @Override
     public List<TopicPartition> partitionsOfBrokers(Set<String> topics, Set<Integer> brokersID) {
       return replicas(topics).entrySet().stream()
@@ -467,7 +499,7 @@ public class Builder {
     }
 
     @Override
-    public void moveTo(Set<Integer> brokers) {
+    public void moveTo(List<Integer> brokers) {
       Utils.handleException(
           () ->
               admin
diff --git a/app/src/main/java/org/astraea/topic/Migrator.java b/app/src/main/java/org/astraea/topic/Migrator.java
index aeec5f19..9a9bfdbd 100644
--- a/app/src/main/java/org/astraea/topic/Migrator.java
+++ b/app/src/main/java/org/astraea/topic/Migrator.java
@@ -1,7 +1,7 @@
 package org.astraea.topic;
 
+import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 /** used to migrate partitions to another broker or broker folder. */
 public interface Migrator {
@@ -24,11 +24,11 @@ public interface Migrator {
 
   /**
    * move partitions to specify brokers. Noted that the number of brokers must be equal to the
-   * number of replicas
+   * number of replicas,and the first broker in the list will be set as the preferred leader
    *
    * @param brokers to host partitions
    */
-  void moveTo(Set<Integer> brokers);
+  void moveTo(List<Integer> brokers);
 
   /**
    * move partition to specify folder.
diff --git a/app/src/main/java/org/astraea/topic/ReplicaCollie.java b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
index 8f702973..4b1960ba 100644
--- a/app/src/main/java/org/astraea/topic/ReplicaCollie.java
+++ b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
@@ -2,9 +2,11 @@ package org.astraea.topic;
 
 import com.beust.jcommander.Parameter;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
@@ -18,8 +20,8 @@ public class ReplicaCollie {
   static final String UNKNOWN = "unknown";
 
   static class MigratorInfo {
-    Set<Integer> brokerSource;
-    Set<Integer> brokerSink;
+    List<Integer> brokerSource;
+    List<Integer> brokerSink;
     Set<String> pathSource;
     Set<String> pathSink;
   }
@@ -39,7 +41,7 @@ public class ReplicaCollie {
       argument.toBrokers =
           admin.brokerIds().stream()
               .filter(b -> !args.fromBrokers.contains(b))
-              .collect(Collectors.toSet());
+              .collect(Collectors.toList());
     } else {
       argument.toBrokers = args.toBrokers.isEmpty() ? args.fromBrokers : args.toBrokers;
     }
@@ -70,7 +72,6 @@ public class ReplicaCollie {
               + topics.iterator().next()
               + " does not exist partition: "
               + args.partitions.toString());
-
     if (args.fromBrokers.containsAll(admin.brokerIds()))
       throw new IllegalArgumentException(
           "No enough available brokers!" + " removed at least one: " + args.fromBrokers);
@@ -87,9 +88,9 @@ public class ReplicaCollie {
     }
   }
 
-  static TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> checkMigratorBroker(
+  static TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> checkMigratorBroker(
       TopicAdmin admin, Argument argument) {
-    TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate =
+    TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate =
         new TreeMap<>(
             Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
     admin.replicas(argument.topics).entrySet().stream()
@@ -102,7 +103,7 @@ public class ReplicaCollie {
         .forEach(
             (tp) -> {
               var currentBrokers =
-                  tp.getValue().stream().map(Replica::broker).collect(Collectors.toSet());
+                  tp.getValue().stream().map(Replica::broker).collect(Collectors.toList());
               var keptBrokers =
                   currentBrokers.stream()
                       .filter(i -> !argument.fromBrokers.contains(i))
@@ -112,7 +113,7 @@ public class ReplicaCollie {
                 var availableBrokers =
                     argument.toBrokers.stream()
                         .filter(i -> !keptBrokers.contains(i) && !argument.fromBrokers.contains(i))
-                        .collect(Collectors.toSet());
+                        .collect(Collectors.toList());
                 if (availableBrokers.size() < numberOfMigratedReplicas)
                   throw new IllegalArgumentException(
                       "No enough available brokers! Available: "
@@ -121,7 +122,15 @@ public class ReplicaCollie {
                           + currentBrokers
                           + " removed: "
                           + argument.fromBrokers);
-                brokerMigrate.put(tp.getKey(), Map.entry(currentBrokers, argument.toBrokers));
+                var targetBrokers = new ArrayList<>(keptBrokers);
+                if (numberOfMigratedReplicas < argument.toBrokers.size())
+                  targetBrokers.addAll(
+                      new ArrayList<>(availableBrokers)
+                          .subList(0, argument.fromBrokers.size() - keptBrokers.size()));
+                else
+                  targetBrokers.addAll(
+                      new ArrayList<>(availableBrokers).subList(0, argument.toBrokers.size()));
+                brokerMigrate.put(tp.getKey(), Map.entry(argument.fromBrokers, targetBrokers));
               } else {
                 brokerMigrate.put(tp.getKey(), Map.entry(currentBrokers, currentBrokers));
               }
@@ -176,7 +185,7 @@ public class ReplicaCollie {
   }
 
   static void brokerMigrator(
-      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate,
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate,
       TopicAdmin admin) {
     brokerMigrate.forEach(
         (tp, assignments) -> {
@@ -201,7 +210,7 @@ public class ReplicaCollie {
   }
 
   static TreeMap<TopicPartition, MigratorInfo> getResult(
-      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate,
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate,
       TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate,
       Argument argument,
       TopicAdmin admin) {
@@ -223,8 +232,8 @@ public class ReplicaCollie {
         });
     pathMigrate.forEach(
         (tp, assignments) -> {
-          Set<Integer> fromBroker;
-          Set<Integer> toBroker;
+          List<Integer> fromBroker;
+          List<Integer> toBroker;
           if (argument.fromBrokers.containsAll(argument.toBrokers)) {
             fromBroker = argument.fromBrokers;
             toBroker = fromBroker;
@@ -241,15 +250,24 @@ public class ReplicaCollie {
             result.put(tp, migratorInfo);
           } else {
             if (assignments.getValue().contains(UNKNOWN)) {
-              Set<String> newPath = new HashSet<>();
-              newPath =
-                  !argument.verify
-                      ? admin.replicas(argument.topics).get(tp).stream()
-                          .map(Replica::path)
-                          .filter(path -> !assignments.getKey().contains(path))
-                          .collect(Collectors.toSet())
-                      : Set.of(UNKNOWN);
-              if (assignments.getKey().equals(newPath)) {
+              Set<String> newPath;
+              if (!argument.verify) {
+                var replicas = admin.replicas(argument.topics).get(tp);
+                newPath =
+                    replicas.stream()
+                            .map(Replica::path)
+                            .filter(p -> !assignments.getKey().contains(p))
+                            .collect(Collectors.toSet())
+                            .isEmpty()
+                        ? replicas.stream().map(Replica::path).collect(Collectors.toSet())
+                        : replicas.stream()
+                            .map(Replica::path)
+                            .filter(p -> !assignments.getKey().contains(p))
+                            .collect(Collectors.toSet());
+              } else {
+                newPath = Set.of(UNKNOWN);
+              }
+              if (assignments.getKey().equals(newPath) && fromBroker == toBroker) {
                 if (!argument.verify) result.remove(tp);
               } else {
                 MigratorInfo migratorInfo = new MigratorInfo();
@@ -279,6 +297,7 @@ public class ReplicaCollie {
 
   public static void main(String[] args) throws IOException {
     var argument = org.astraea.argument.Argument.parse(new Argument(), args);
+
     try (var admin = TopicAdmin.of(argument.props())) {
       execute(admin, argument)
           .forEach(
@@ -310,17 +329,13 @@ public class ReplicaCollie {
     @Parameter(
         names = {"--from"},
         description = "Those brokers won't hold any replicas of topics (defined by --topics)",
-        validateWith = IntegerSetField.class,
-        converter = IntegerSetField.class,
         required = true)
-    Set<Integer> fromBrokers;
+    List<Integer> fromBrokers;
 
     @Parameter(
         names = {"--to"},
-        description = "The replicas of topics (defined by --topic) will be moved to those brokers",
-        validateWith = IntegerSetField.class,
-        converter = IntegerSetField.class)
-    Set<Integer> toBrokers = Collections.emptySet();
+        description = "The replicas of topics (defined by --topic) will be moved to those brokers")
+    List<Integer> toBrokers = List.of();
 
     @Parameter(
         names = {"--partitions"},
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index 4f944611..c39c516b 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -1,6 +1,7 @@
 package org.astraea.topic;
 
 import java.io.Closeable;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -59,6 +60,12 @@ public interface TopicAdmin extends Closeable {
   /** @return all brokers' ids */
   Set<Integer> brokerIds();
 
+  /**
+   * @parm partitions map of TopicPartition and target brokers
+   * @return true if the leader is change successful changed
+   */
+  Map<TopicPartition, Boolean> changeReplicaLeader(Map<TopicPartition, Integer> partitions);
+
   /**
    * @param topics topic names
    * @param brokersID brokers ID
@@ -70,7 +77,7 @@ public interface TopicAdmin extends Closeable {
    * @param brokers a Set containing broker's ID
    * @return all log directory
    */
-  Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers);
+  Map<Integer, Set<String>> brokerFolders(Collection<Integer> brokers);
 
   /** @return a partition migrator used to move partitions to another broker or folder. */
   Migrator migrator();


Test patch:
diff --git a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
index 729d3cb0..8679633a 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.io.IOException;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
@@ -48,14 +49,14 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
       var brokerSource =
           partitionReplicas.get(new TopicPartition(topicName, 0)).stream()
               .map(Replica::broker)
-              .collect(Collectors.toSet());
+              .collect(Collectors.toList());
       var brokerSink =
           topicAdmin.brokerIds().stream().filter(b -> !brokerSource.contains(b)).iterator().next();
-      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate =
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate =
           new TreeMap<>(
               Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
       brokerMigrate.put(
-          new TopicPartition(topicName, 0), Map.entry(brokerSource, Set.of(brokerSink)));
+          new TopicPartition(topicName, 0), Map.entry(brokerSource, List.of(brokerSink)));
       Assertions.assertEquals(
           topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).size(), 2);
       ReplicaCollie.brokerMigrator(brokerMigrate, topicAdmin);
@@ -135,14 +136,14 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
               .filter(Replica::isCurrent)
               .collect(Collectors.toList());
       Assertions.assertEquals(1, replicas.size());
-      var badBroker = replicas.stream().map(Replica::broker).collect(Collectors.toSet());
+      var badBroker = replicas.stream().map(Replica::broker).collect(Collectors.toList());
       var targetBroker =
           topicAdmin.brokerIds().stream()
               .filter(b -> !badBroker.contains(b))
-              .collect(Collectors.toSet());
+              .collect(Collectors.toList());
       var argument = new ReplicaCollie.Argument();
       argument.fromBrokers = badBroker;
-      argument.toBrokers = targetBroker;
+      argument.toBrokers = targetBroker.subList(0, 1);
       argument.brokers = bootstrapServers();
       argument.topics = Set.of(topicName);
       argument.partitions = Set.of(0);
@@ -161,8 +162,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
               for (var index = 0; index != rs.size(); ++index)
                 Assertions.assertEquals(rs.get(index), currentRs.get(index));
             });
-        Assertions.assertEquals(targetBroker, assignment.brokerSink);
-
+        Assertions.assertEquals(argument.toBrokers, assignment.brokerSink);
       } else {
         Utils.waitFor(
             () -> {
diff --git a/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java b/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java
index 8685ea17..a694460a 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java
@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -50,7 +51,7 @@ class ReplicaSyncingMonitorIntegrationTest extends RequireBrokerCluster {
       Thread executionThread =
           new Thread(
               () -> {
-                topicAdmin.migrator().partition(TOPIC_NAME, 0).moveTo(Set.of(moveToBroker));
+                topicAdmin.migrator().partition(TOPIC_NAME, 0).moveTo(List.of(moveToBroker));
                 ReplicaSyncingMonitor.execute(
                     topicAdmin,
                     org.astraea.argument.Argument.parse(
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index 463a8495..4f29ce77 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -154,6 +155,8 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
+  // There is a problem when migrating the log folder under Windows because the migrated source
+  // cannot be deleted, so disabled this test on Windows for now.
   @DisabledOnOs(WINDOWS)
   void testMigrateSinglePartition() throws InterruptedException {
     var topicName = "testMigrateSinglePartition";
@@ -162,7 +165,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
       var broker = topicAdmin.brokerIds().iterator().next();
-      topicAdmin.migrator().partition(topicName, 0).moveTo(Set.of(broker));
+      topicAdmin.migrator().partition(topicName, 0).moveTo(List.of(broker));
       Utils.waitFor(
           () -> {
             var replicas = topicAdmin.replicas(Set.of(topicName));
@@ -205,6 +208,38 @@ public class TopicAdminTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void testChangeReplicaLeader() throws InterruptedException {
+    var topicName = "testChangeReplicaLeader";
+    var tp = new TopicPartition(topicName, 0);
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin
+          .creator()
+          .topic(topicName)
+          .numberOfPartitions(1)
+          .numberOfReplicas((short) 2)
+          .create();
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
+      topicAdmin.replicas(Set.of(topicName)).get(tp);
+      var replicas = topicAdmin.replicas(Set.of(topicName)).get(tp);
+      var oldLeader =
+          replicas.stream().filter(Replica::leader).collect(Collectors.toList()).get(0).broker();
+      var newLeader =
+          replicas.stream().filter(b -> !b.leader()).collect(Collectors.toList()).get(0).broker();
+      topicAdmin.changeReplicaLeader(Map.of(tp, newLeader));
+      Assertions.assertNotEquals(oldLeader, newLeader);
+      Utils.waitFor(
+          () ->
+              topicAdmin.replicas(Set.of(topicName)).get(tp).stream()
+                  .filter(r -> r.broker() == newLeader)
+                  .iterator()
+                  .next()
+                  .leader());
+    }
+  }
+
   @Test
   @DisabledOnOs(WINDOWS)
   void testMigrateAllPartitions() throws InterruptedException {
@@ -214,7 +249,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
       var broker = topicAdmin.brokerIds().iterator().next();
-      topicAdmin.migrator().topic(topicName).moveTo(Set.of(broker));
+      topicAdmin.migrator().topic(topicName).moveTo(List.of(broker));
       Utils.waitFor(
           () -> {
             var replicas = topicAdmin.replicas(Set.of(topicName));