Repository: skiptests/astraea
Pull request #203 (2022-01-28 04:10:59)

Improvements to problems in testing
針對測試過程中產生的問題進行一些修復。
#196 中出現的問題會有一定概率同樣發生在performance創建過程中，以同樣的方式wait()進行處理。
docker2G的內存有一些影響我們的performance的測試，現在調整為12G。

chia7712: > docker2G的內存有一些影響我們的performance的測試，現在調整為12G。

這是禮拜三討論到的問題嗎？可否描述清楚一點

Patch:
diff --git a/README.md b/README.md
index 286e0cb0..ecdcf90f 100644
--- a/README.md
+++ b/README.md
@@ -295,24 +295,47 @@ java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --
 
 ## Replica Collie
 
-This tool offers an effective way to migrate all replicas from specific brokers to others.
+This tool offers an effective way to migrate specify replicas from specific brokers to others brokers.
 
 ### Move all replicas from broker_0 and broker_1 to other brokers
 
-```shell
+```shell 
 ./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0,1"
 ```
 
 ### 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 --to 1 --topics abc"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0,1 --to 2,3 --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"
+```
+
+### 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"
+```
+
+### 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"
 ```
 
 ### Replica Collie Configurations
+
 1. --bootstrap.servers: the server to connect to
-2. --topics: the topics to be moved
-3. --admin.props.file: the file path containing the properties to be passed to kafka admin
+2. --from: the broker of the specified replica
+3. --prop.file: the file path containing the properties to be passed to kafka admin
+4. --to: the replica moved target broker
+5. --topics: the topics to be moved
+6. --partitions : all partitions that will be moved
+7. --path: the replica that will move to
 
 ## Kafka Partition Score
 
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
deleted file mode 100644
index 5667d1a7..00000000
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package org.astraea.partitioner.smoothPartitioner;
-
-/**
- * The interface for the {@link SmoothWeightPartitioner}
- *
- * @see SmoothWeightPartitioner Enable users to send dependent data.
- */
-public interface DependencyClient {
-  void startDependency();
-
-  void finishDependency();
-
-  String state();
-}
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
index 287af45e..76585318 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
@@ -9,11 +9,8 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicReference;
-import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.KafkaException;
-import org.astraea.Utils;
 import org.astraea.partitioner.nodeLoadMetric.LoadPoisson;
 import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
 
@@ -21,24 +18,8 @@ import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
  * Based on the jmx metrics obtained from Kafka, it records the load status of the node over a
  * period of time. Predict the future status of each node through the poisson of the load status.
  * Finally, the result of poisson is used as the weight to perform smooth weighted RoundRobin.
- *
- * <p>SmoothWeightPartitioner also allows users to make producers send dependency data.
- *
- * <pre>{@code
- * KafkaProducer producer = new KafkaProducer(props);
- *
- * var dependencyControl = SmoothWeightPartitioner.dependencyClient(producer);
- * dependencyControl.startDependency();
- * try{
- *     producer.send();
- *     producer.send();
- *     producer.send();
- * } finally{
- *     dependencyControl.finishDependency();
- * }
- * }</pre>
  */
-public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
+public class SmoothWeightPartitioner implements Partitioner {
 
   /**
    * Record the current weight of each node according to Poisson calculation and the weight after
@@ -47,67 +28,46 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
   private final Map<Integer, int[]> brokersWeight = new HashMap<>();
 
   private NodeLoadClient nodeLoadClient;
-  private DependencyManager dependencyManager;
-  /**
-   * This parameter only works in dependency mode.And it will be specified when the dependency is
-   * executed for the first time, and no changes will be made afterwards.
-   */
-  private int targetPartition;
 
   @Override
   public int partition(
       String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
     Map<Integer, Integer> loadCount;
     var rand = new Random();
-    switch (dependencyManager.currentState) {
-      case START_DEPENDENCY:
-        loadCount = loadCount(cluster);
-        Objects.requireNonNull(loadCount, "OverLoadCount should not be null.");
-        var minOverLoadNode = loadCount.entrySet().stream().min(Map.Entry.comparingByValue()).get();
-        var minOverLoadPartition = cluster.partitionsForNode(minOverLoadNode.getKey());
-        targetPartition =
-            minOverLoadPartition.get(rand.nextInt(minOverLoadPartition.size())).partition();
-        dependencyManager.transitionTo(DependencyManager.State.IN_DEPENDENCY);
-        return targetPartition;
-      case IN_DEPENDENCY:
-        return targetPartition;
-      default:
-        loadCount = loadCount(cluster);
-        var loadPoisson = new LoadPoisson();
-
-        Objects.requireNonNull(loadCount, "OverLoadCount should not be null.");
-        brokersWeight(loadPoisson.allPoisson(loadCount));
-        AtomicReference<Map.Entry<Integer, int[]>> maxWeightServer = new AtomicReference<>();
-        var allWeight = allNodesWeight();
-        brokersWeight.forEach(
-            (nodeID, weight) -> {
-              if (maxWeightServer.get() == null
-                  || weight[1] > maxWeightServer.get().getValue()[1]) {
-                maxWeightServer.set(Map.entry(nodeID, weight));
+    loadCount = loadCount(cluster);
+    var loadPoisson = new LoadPoisson();
+
+    Objects.requireNonNull(loadCount, "OverLoadCount should not be null.");
+    brokersWeight(loadPoisson.allPoisson(loadCount));
+    AtomicReference<Map.Entry<Integer, int[]>> maxWeightServer = new AtomicReference<>();
+    var allWeight = allNodesWeight();
+    brokersWeight.forEach(
+        (nodeID, weight) -> {
+          if (maxWeightServer.get() == null || weight[1] > maxWeightServer.get().getValue()[1]) {
+            maxWeightServer.set(Map.entry(nodeID, weight));
+          }
+          brokersWeight.put(nodeID, new int[] {weight[0], weight[1] + weight[0]});
+        });
+    Objects.requireNonNull(maxWeightServer.get(), "MaxWeightServer should not be null.");
+    brokersWeight.put(
+        maxWeightServer.get().getKey(),
+        new int[] {
+          maxWeightServer.get().getValue()[0], maxWeightServer.get().getValue()[1] - allWeight
+        });
+    brokersWeight
+        .keySet()
+        .forEach(
+            brokerID -> {
+              if (memoryWarning(brokerID)) {
+                subBrokerWeight(brokerID, 100);
               }
-              brokersWeight.put(nodeID, new int[] {weight[0], weight[1] + weight[0]});
-            });
-        Objects.requireNonNull(maxWeightServer.get(), "MaxWeightServer should not be null.");
-        brokersWeight.put(
-            maxWeightServer.get().getKey(),
-            new int[] {
-              maxWeightServer.get().getValue()[0], maxWeightServer.get().getValue()[1] - allWeight
             });
-        brokersWeight
-            .keySet()
-            .forEach(
-                brokerID -> {
-                  if (memoryWarning(brokerID)) {
-                    subBrokerWeight(brokerID, 100);
-                  }
-                });
 
-        ArrayList<Integer> partitionList = new ArrayList<>();
-        cluster
-            .partitionsForNode(maxWeightServer.get().getKey())
-            .forEach(partitionInfo -> partitionList.add(partitionInfo.partition()));
-        return partitionList.get(rand.nextInt(partitionList.size()));
-    }
+    ArrayList<Integer> partitionList = new ArrayList<>();
+    cluster
+        .partitionsForNode(maxWeightServer.get().getKey())
+        .forEach(partitionInfo -> partitionList.add(partitionInfo.partition()));
+    return partitionList.get(rand.nextInt(partitionList.size()));
   }
 
   @Override
@@ -131,7 +91,6 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
       Objects.requireNonNull(mapAddress, "You must configure jmx_servers correctly.");
 
       nodeLoadClient = new NodeLoadClient(mapAddress);
-      dependencyManager = new DependencyManager();
     } catch (IOException e) {
       throw new RuntimeException();
     }
@@ -163,20 +122,6 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
     return brokersWeight.values().stream().mapToInt(vs -> vs[0]).sum();
   }
 
-  public static DependencyClient dependencyClient(KafkaProducer<?, ?> producer) {
-    return (DependencyClient) Utils.requireField(producer, "partitioner");
-  }
-
-  @Override
-  public synchronized void startDependency() {
-    dependencyManager.startDependency();
-  }
-
-  @Override
-  public synchronized void finishDependency() {
-    dependencyManager.finishDependency();
-  }
-
   private boolean memoryWarning(int brokerID) {
     return nodeLoadClient.memoryUsage(brokerID) >= 0.8;
   }
@@ -188,60 +133,4 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
       throw new IllegalArgumentException(e);
     }
   }
-
-  // visible of testing
-  public String state() {
-    return dependencyManager.currentState.toString();
-  }
-
-  /**
-   * Store the state related to the partitioner dependency and avoid wrong transitions between them.
-   */
-  private static class DependencyManager {
-    private volatile State currentState = State.UNINITIALIZED;
-
-    private synchronized void startDependency() {
-      transitionTo(State.START_DEPENDENCY);
-    }
-
-    private synchronized void finishDependency() {
-      transitionTo(State.UNINITIALIZED);
-    }
-
-    private enum State {
-      UNINITIALIZED,
-      START_DEPENDENCY,
-      IN_DEPENDENCY,
-      FATAL_ERROR;
-
-      private boolean isTransitionValid(
-          DependencyManager.State source, DependencyManager.State target) {
-        switch (target) {
-          case UNINITIALIZED:
-            return source == START_DEPENDENCY || source == IN_DEPENDENCY;
-          case START_DEPENDENCY:
-            return source == UNINITIALIZED;
-          case IN_DEPENDENCY:
-            return source == START_DEPENDENCY;
-          default:
-            return true;
-        }
-      }
-    }
-
-    private void transitionTo(State target) {
-      try {
-        if (!currentState.isTransitionValid(currentState, target)) {
-          throw new KafkaException(
-              "Invalid transition attempted from state "
-                  + currentState.name()
-                  + " to state "
-                  + target.name());
-        }
-      } catch (KafkaException e) {
-        // TODO
-      }
-      currentState = target;
-    }
-  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Distribution.java b/app/src/main/java/org/astraea/performance/Distribution.java
index 8148aad6..bbfc5955 100644
--- a/app/src/main/java/org/astraea/performance/Distribution.java
+++ b/app/src/main/java/org/astraea/performance/Distribution.java
@@ -19,6 +19,8 @@ public interface Distribution {
       var name = rawArgument.split(":")[0];
 
       switch (name) {
+        case "fixed":
+          return fixed();
         case "uniform":
           if (args.length > 1) return uniform(Integer.parseInt(args[1]));
           return uniform();
@@ -34,6 +36,15 @@ public interface Distribution {
     }
   }
 
+  static Distribution fixed() {
+    var rand = new Random();
+    return fixed(rand.nextLong());
+  }
+
+  static Distribution fixed(long value) {
+    return () -> value;
+  }
+
   /** A distribution for providing a random long number from range [0, 2147483647) */
   static Distribution uniform() {
     return uniform(Integer.MAX_VALUE);
diff --git a/app/src/main/java/org/astraea/performance/FileWriter.java b/app/src/main/java/org/astraea/performance/FileWriter.java
index f59c00a8..f29e1756 100644
--- a/app/src/main/java/org/astraea/performance/FileWriter.java
+++ b/app/src/main/java/org/astraea/performance/FileWriter.java
@@ -6,6 +6,7 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.stream.IntStream;
 import org.astraea.concurrent.ThreadPool;
+import org.astraea.utils.DataUnit;
 
 public class FileWriter implements ThreadPool.Executor {
   private final Manager manager;
@@ -35,7 +36,7 @@ public class FileWriter implements ThreadPool.Executor {
     try {
       writer = new BufferedWriter(new java.io.FileWriter(CSVName));
       writer.write(
-          "Time \\ Name, Consumed/Produced, Output throughput (MiB/sec), Input throughput (MiB/sec), "
+          "Time \\ Name, Consumed/Produced, Output throughput (/sec), Input throughput (/sec), "
               + "Publish max latency (ms), Publish min latency (ms), "
               + "End-to-end max latency (ms), End-to-end min latency (ms)");
       IntStream.range(0, tracker.producerResult().bytes.size())
@@ -45,9 +46,9 @@ public class FileWriter implements ThreadPool.Executor {
                   writer.write(
                       ",Producer["
                           + i
-                          + "] average throughput (MB/sec), Producer["
+                          + "] current throughput (/sec), Producer["
                           + i
-                          + "] average publish latency (ms)");
+                          + "] current publish latency (ms)");
                 } catch (IOException ignore) {
                 }
               });
@@ -58,9 +59,9 @@ public class FileWriter implements ThreadPool.Executor {
                   writer.write(
                       ",Consumer["
                           + i
-                          + "] average throughput (MB/sec), Consumer["
+                          + "] current throughput (/sec), Consumer["
                           + i
-                          + "] average ene-to-end latency (ms)");
+                          + "] current ene-to-end latency (ms)");
                 } catch (IOException ignore) {
                 }
               });
@@ -89,16 +90,16 @@ public class FileWriter implements ThreadPool.Executor {
               + duration.toSecondsPart()
               + "s");
       writer.write(String.format(",%.2f%% / %.2f%%", consumerPercentage, producerPercentage));
-      writer.write("," + producerResult.averageBytes(duration));
-      writer.write("," + consumerResult.averageBytes(duration));
+      writer.write("," + DataUnit.Byte.of(producerResult.totalCurrentBytes()));
+      writer.write("," + DataUnit.Byte.of(consumerResult.totalCurrentBytes()));
       writer.write("," + producerResult.maxLatency + "," + producerResult.minLatency);
       writer.write("," + consumerResult.maxLatency + "," + consumerResult.minLatency);
       for (int i = 0; i < producerResult.bytes.size(); ++i) {
-        writer.write("," + Tracker.avg(duration, producerResult.bytes.get(i)));
+        writer.write("," + DataUnit.Byte.of(producerResult.currentBytes.get(i)));
         writer.write("," + producerResult.averageLatencies.get(i));
       }
       for (int i = 0; i < consumerResult.bytes.size(); ++i) {
-        writer.write("," + Tracker.avg(duration, consumerResult.bytes.get(i)));
+        writer.write("," + DataUnit.Byte.of(consumerResult.currentBytes.get(i)));
         writer.write("," + consumerResult.averageLatencies.get(i));
       }
       writer.newLine();
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index 2fb59b53..0b5fdec6 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -100,13 +100,13 @@ public class Manager {
 
   /** Check if we should keep consuming record. */
   public boolean consumedDone() {
-    return producedDone() && consumedRecords() >= producedRecords();
+    return producedDone()
+        && (consumerMetrics.size() == 0 || consumedRecords() >= producedRecords());
   }
 
   /** Randomly choose a key according to the distribution. */
-  public Optional<byte[]> getKey() {
-    if (distribution == null) return Optional.empty();
-    return Optional.of((String.valueOf(distribution.get())).getBytes());
+  public byte[] getKey() {
+    return (String.valueOf(distribution.get())).getBytes();
   }
 
   /** Randomly generate content before {@link #getContent()} is called. */
diff --git a/app/src/main/java/org/astraea/performance/Metrics.java b/app/src/main/java/org/astraea/performance/Metrics.java
index ec279c4f..90ebedbc 100644
--- a/app/src/main/java/org/astraea/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/performance/Metrics.java
@@ -9,6 +9,7 @@ public class Metrics implements BiConsumer<Long, Long> {
   private long max;
   private long min;
   private long bytes;
+  private long currentBytes;
 
   public Metrics() {
     avgLatency = 0;
@@ -16,6 +17,7 @@ public class Metrics implements BiConsumer<Long, Long> {
     max = 0;
     min = Long.MAX_VALUE;
     bytes = 0;
+    currentBytes = 0;
   }
 
   /** Simultaneously add latency and bytes. */
@@ -33,6 +35,7 @@ public class Metrics implements BiConsumer<Long, Long> {
   }
   /** Add a new value to bytes. */
   private synchronized void addBytes(long bytes) {
+    this.currentBytes += bytes;
     this.bytes += bytes;
   }
 
@@ -57,4 +60,10 @@ public class Metrics implements BiConsumer<Long, Long> {
   public synchronized long bytes() {
     return bytes;
   }
+
+  public synchronized long clearAndGetCurrentBytes() {
+    var ans = currentBytes;
+    currentBytes = 0;
+    return ans;
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 5a4cd928..524b5a2b 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -5,7 +5,12 @@ import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import java.io.IOException;
 import java.time.Duration;
-import java.util.*;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -15,8 +20,10 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.record.CompressionType;
+import org.astraea.Utils;
 import org.astraea.argument.ArgumentUtil;
 import org.astraea.argument.BasicArgumentWithPropFile;
 import org.astraea.concurrent.ThreadPool;
@@ -71,6 +78,7 @@ public class Performance {
           .topic(param.topic)
           .create();
 
+      Utils.waitFor(() -> topicAdmin.topicNames().contains(param.topic));
       partitions = partition(param, topicAdmin);
     }
 
@@ -185,7 +193,7 @@ public class Performance {
             .sender()
             .topic(param.topic)
             .partition(partitions.get(rand.nextInt(partitions.size())))
-            .key(manager.getKey().orElse(null))
+            .key(manager.getKey())
             .value(payload.get())
             .timestamp(start)
             .run()
@@ -212,7 +220,7 @@ public class Performance {
       return topicAdmin
           .partitionsOfBrokers(Set.of(param.topic), new HashSet<>(param.specifyBroker))
           .stream()
-          .map(topicPartition -> topicPartition.partition())
+          .map(TopicPartition::partition)
           .collect(Collectors.toList());
     } else return List.of(-1);
   }
@@ -310,9 +318,9 @@ public class Performance {
     @Parameter(
         names = {"--key.distribution"},
         description =
-            "String: Distribution name. Available distribution names: \"uniform\", \"zipfian\", \"latest\". Default: (No key)",
+            "String: Distribution name. Available distribution names: \"fixed\" \"uniform\", \"zipfian\", \"latest\". Default: uniform",
         converter = Distribution.DistributionConverter.class)
-    Distribution distribution = null;
+    Distribution distribution = Distribution.uniform();
 
     @Parameter(
         names = {"--specify.broker"},
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index fbd25def..ab12bc1a 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -3,6 +3,7 @@ package org.astraea.performance;
 import java.time.Duration;
 import java.util.*;
 import org.astraea.concurrent.ThreadPool;
+import org.astraea.utils.DataUnit;
 
 /** Print out the given metrics. */
 public class Tracker implements ThreadPool.Executor {
@@ -60,12 +61,14 @@ public class Tracker implements ThreadPool.Executor {
             + duration.toSecondsPart()
             + "sec");
     System.out.printf("producers completion rate: %.2f%%%n", percentage);
-    System.out.printf("  Throughput: %.3fMB/second%n", result.averageBytes(duration));
+    System.out.printf("  average Throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf(
+        "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
     System.out.println("  publish max latency: " + result.maxLatency + "ms");
     System.out.println("  publish mim latency: " + result.minLatency + "ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
-          "  producer[%d] average throughput: %.3fMB%n", i, avg(duration, result.bytes.get(i)));
+          "  producer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
           "  producer[%d] average publish latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
@@ -82,12 +85,14 @@ public class Tracker implements ThreadPool.Executor {
     // Print out percentage of (consumed records) and (produced records)
     var percentage = result.completedRecords * 100D / manager.producedRecords();
     System.out.printf("consumer completion rate: %.2f%%%n", percentage);
-    System.out.printf("  throughput: %.3fMB/second%n", result.averageBytes(duration));
+    System.out.printf("  average throughput: %.3f MB/second%n", result.averageBytes(duration));
+    System.out.printf(
+        "  current throughput: %s/second%n", DataUnit.Byte.of(result.totalCurrentBytes()));
     System.out.println("  end-to-end max latency: " + result.maxLatency + "ms");
     System.out.println("  end-to-end mim latency: " + result.minLatency + "ms");
     for (int i = 0; i < result.bytes.size(); ++i) {
       System.out.printf(
-          "  consumer[%d] average throughput: %.3fMB%n", i, avg(duration, result.bytes.get(i)));
+          "  consumer[%d] average throughput: %.3f MB%n", i, avg(duration, result.bytes.get(i)));
       System.out.printf(
           "  consumer[%d] average ene-to-end latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
@@ -107,12 +112,14 @@ public class Tracker implements ThreadPool.Executor {
   private static Result result(List<Metrics> metrics) {
     var completed = 0;
     var bytes = new ArrayList<Long>();
+    var currentBytes = new ArrayList<Long>();
     var averageLatencies = new ArrayList<Double>();
     var max = 0L;
     var min = Long.MAX_VALUE;
     for (Metrics data : metrics) {
       completed += data.num();
       bytes.add(data.bytes());
+      currentBytes.add(data.clearAndGetCurrentBytes());
       averageLatencies.add(data.avgLatency());
       max = Math.max(max, data.max());
       min = Math.min(min, data.min());
@@ -120,6 +127,7 @@ public class Tracker implements ThreadPool.Executor {
     return new Result(
         completed,
         Collections.unmodifiableList(bytes),
+        Collections.unmodifiableList(currentBytes),
         Collections.unmodifiableList(averageLatencies),
         min,
         max);
@@ -128,6 +136,7 @@ public class Tracker implements ThreadPool.Executor {
   static class Result {
     public final long completedRecords;
     public final List<Long> bytes;
+    public final List<Long> currentBytes;
     public final List<Double> averageLatencies;
     public final long minLatency;
     public final long maxLatency;
@@ -135,11 +144,13 @@ public class Tracker implements ThreadPool.Executor {
     Result(
         long completedRecords,
         List<Long> bytes,
+        List<Long> currentBytes,
         List<Double> averageLatencies,
         long minLatency,
         long maxLatency) {
       this.completedRecords = completedRecords;
       this.bytes = bytes;
+      this.currentBytes = currentBytes;
       this.averageLatencies = averageLatencies;
       this.minLatency = minLatency;
       this.maxLatency = maxLatency;
@@ -152,5 +163,9 @@ public class Tracker implements ThreadPool.Executor {
     long totalBytes() {
       return bytes.stream().mapToLong(i -> i).sum();
     }
+
+    long totalCurrentBytes() {
+      return currentBytes.stream().mapToLong(i -> i).sum();
+    }
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/ReplicaCollie.java b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
index 6abca971..dcc21f45 100644
--- a/app/src/main/java/org/astraea/topic/ReplicaCollie.java
+++ b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
@@ -9,62 +9,268 @@ import org.astraea.argument.ArgumentUtil;
 import org.astraea.argument.BasicArgumentWithPropFile;
 
 public class ReplicaCollie {
+  static final String UNKNOWN = "unknown";
 
-  static Map<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> execute(
-      TopicAdmin admin, Argument args) {
+  static class MigratorInfo {
+    Set<Integer> brokerSource;
+    Set<Integer> brokerSink;
+    Set<String> pathSource;
+    Set<String> pathSink;
+  }
+
+  static Argument setArguments(TopicAdmin admin, Argument args) {
+    Argument argument = new Argument();
+    argument.topics = args.topics.isEmpty() ? admin.topicNames() : args.topics;
+    argument.fromBrokers = args.fromBrokers;
+    argument.partitions = args.partitions;
+    argument.path = args.path.isEmpty() ? null : args.path;
+    argument.verify = args.verify;
+    // If the partitions and the broker of to migrate path are not specified
+    // at the same time, the topics specified by the broker will be moved to
+    // other brokers (if not specified, all topics of the broker will be
+    // included).
+    if (args.partitions.isEmpty() && args.path.isEmpty() && args.toBrokers.isEmpty()) {
+      argument.toBrokers =
+          admin.brokerIds().stream()
+              .filter(b -> !args.fromBrokers.contains(b))
+              .collect(Collectors.toSet());
+    } else {
+      argument.toBrokers = args.toBrokers.isEmpty() ? args.fromBrokers : args.toBrokers;
+    }
+    return argument;
+  }
+
+  static void checkArgs(TopicAdmin admin, Argument args) {
     var topics = args.topics.isEmpty() ? admin.topicNames() : args.topics;
-    var allBrokers = admin.brokerIds();
-    if (!args.toBrokers.isEmpty() && !allBrokers.containsAll(args.toBrokers))
+    var path = args.path.isEmpty() ? null : args.path.iterator().next();
+    var targetBrokers = args.toBrokers.isEmpty() ? args.fromBrokers : args.toBrokers;
+    var targetPaths = admin.brokerFolders(admin.brokerIds()).get(targetBrokers.iterator().next());
+
+    if (!args.toBrokers.isEmpty() && !admin.brokerIds().containsAll(args.toBrokers))
       throw new IllegalArgumentException(
           "those brokers: "
               + args.toBrokers.stream()
-                  .filter(i -> !allBrokers.contains(i))
+                  .filter(i -> !admin.brokerIds().contains(i))
                   .map(String::valueOf)
                   .collect(Collectors.joining(","))
               + " are nonexistent");
+    if (!args.partitions.isEmpty()
+        && !admin.replicas(topics).keySet().stream()
+            .map(TopicPartition::partition)
+            .collect(Collectors.toSet())
+            .containsAll(args.partitions))
+      throw new IllegalArgumentException(
+          "Topic "
+              + topics.iterator().next()
+              + " does not exist partition: "
+              + args.partitions.toString());
 
-    var targetBrokers = args.toBrokers.isEmpty() ? allBrokers : args.toBrokers;
-    var result =
-        new TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>>(
+    if (args.fromBrokers.containsAll(admin.brokerIds()))
+      throw new IllegalArgumentException(
+          "No enough available brokers!" + " removed at least one: " + args.fromBrokers);
+    if (args.topics.size() >= 2 && !args.partitions.isEmpty())
+      throw new IllegalArgumentException(
+          "When specifying multiple topics, --partitions cannot be specified.");
+    if (path != null) {
+      if (!args.toBrokers.isEmpty() && !targetPaths.contains(path))
+        throw new IllegalArgumentException(
+            "path: " + path + " is not in broker" + args.toBrokers.iterator().next());
+      if (args.toBrokers.isEmpty() && !targetPaths.contains(path))
+        throw new IllegalArgumentException(
+            "path: " + path + " is not in broker" + args.fromBrokers.iterator().next());
+    }
+  }
+
+  static TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> checkMigratorBroker(
+      TopicAdmin admin, Argument argument) {
+    TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate =
+        new TreeMap<>(
             Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
-    admin
-        .replicas(topics)
+    admin.replicas(argument.topics).entrySet().stream()
+        .filter(
+            tp ->
+                ((argument.partitions.contains(tp.getKey().partition()))
+                        || argument.partitions.isEmpty())
+                    && !argument.fromBrokers.containsAll(argument.toBrokers))
+        .collect(Collectors.toSet())
         .forEach(
-            (tp, replicas) -> {
+            (tp) -> {
               var currentBrokers =
-                  replicas.stream().map(Replica::broker).collect(Collectors.toSet());
+                  tp.getValue().stream().map(Replica::broker).collect(Collectors.toSet());
               var keptBrokers =
                   currentBrokers.stream()
-                      .filter(i -> !args.fromBrokers.contains(i))
+                      .filter(i -> !argument.fromBrokers.contains(i))
                       .collect(Collectors.toSet());
               var numberOfMigratedReplicas = currentBrokers.size() - keptBrokers.size();
               if (numberOfMigratedReplicas > 0) {
                 var availableBrokers =
-                    targetBrokers.stream()
-                        .filter(i -> !keptBrokers.contains(i) && !args.fromBrokers.contains(i))
+                    argument.toBrokers.stream()
+                        .filter(i -> !keptBrokers.contains(i) && !argument.fromBrokers.contains(i))
                         .collect(Collectors.toSet());
                 if (availableBrokers.size() < numberOfMigratedReplicas)
                   throw new IllegalArgumentException(
                       "No enough available brokers! Available: "
-                          + targetBrokers
+                          + argument.toBrokers
                           + " current: "
                           + currentBrokers
                           + " removed: "
-                          + args.fromBrokers);
-                var finalBrokers = new HashSet<>(keptBrokers);
-                finalBrokers.addAll(
-                    new ArrayList<>(availableBrokers).subList(0, numberOfMigratedReplicas));
-                result.put(tp, Map.entry(currentBrokers, finalBrokers));
+                          + argument.fromBrokers);
+                brokerMigrate.put(tp.getKey(), Map.entry(currentBrokers, argument.toBrokers));
+              } else {
+                brokerMigrate.put(tp.getKey(), Map.entry(currentBrokers, currentBrokers));
               }
             });
-    result.forEach(
+    return brokerMigrate;
+  }
+
+  static TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> checkMigratorPath(
+      TopicAdmin admin, Argument argument) {
+    TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate =
+        new TreeMap<>(
+            Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
+    admin.replicas(argument.topics).entrySet().stream()
+        .filter(
+            tp ->
+                argument.topics.contains(tp.getKey().topic())
+                    && ((argument.partitions.isEmpty())
+                        || argument.partitions.contains(tp.getKey().partition())))
+        .collect(Collectors.toList())
+        .forEach(
+            (tp) -> {
+              Set<String> fromPath = new HashSet<>();
+              Set<String> toPath = new HashSet<>();
+              tp.getValue().stream()
+                  .filter(assignments -> argument.fromBrokers.contains(assignments.broker()))
+                  .forEach(
+                      assignments -> {
+                        var currentPath = assignments.path();
+                        fromPath.add(currentPath);
+                        if (argument.path != null) {
+                          if (!currentPath.equals(argument.path.iterator().next()))
+                            toPath.add(argument.path.iterator().next());
+                        } else {
+                          if (argument.fromBrokers.contains(argument.toBrokers.iterator().next())) {
+                            toPath.add(
+                                admin
+                                    .brokerFolders(argument.toBrokers)
+                                    .get(argument.toBrokers.iterator().next())
+                                    .stream()
+                                    .filter(p -> !currentPath.contains(p))
+                                    .collect(Collectors.toSet())
+                                    .iterator()
+                                    .next());
+                          } else {
+                            toPath.add(UNKNOWN);
+                          }
+                        }
+                      });
+              if (!fromPath.isEmpty()) pathMigrate.put(tp.getKey(), Map.entry(fromPath, toPath));
+            });
+    return pathMigrate;
+  }
+
+  static void brokerMigrator(
+      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate,
+      TopicAdmin admin) {
+    brokerMigrate.forEach(
         (tp, assignments) -> {
-          if (!args.verify)
+          if (!assignments.getKey().equals(assignments.getValue())) {
             admin.migrator().partition(tp.topic(), tp.partition()).moveTo(assignments.getValue());
+          }
+        });
+  }
+
+  static void pathMigrator(
+      TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate,
+      TopicAdmin admin,
+      Integer broker) {
+    pathMigrate.forEach(
+        (tp, assignments) -> {
+          if (assignments.getValue().size() > 0)
+            admin
+                .migrator()
+                .partition(tp.topic(), tp.partition())
+                .moveTo(Map.of(broker, assignments.getValue().iterator().next()));
+        });
+  }
+
+  static TreeMap<TopicPartition, MigratorInfo> getResult(
+      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate,
+      TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate,
+      Argument argument,
+      TopicAdmin admin) {
+    var result =
+        new TreeMap<TopicPartition, MigratorInfo>(
+            Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
+    brokerMigrate.forEach(
+        (tp, assignments) -> {
+          if (!assignments.getKey().equals(assignments.getValue())) {
+            if (!pathMigrate.isEmpty()) {
+              MigratorInfo migratorInfo = new MigratorInfo();
+              migratorInfo.brokerSource = assignments.getKey();
+              migratorInfo.brokerSink = assignments.getValue();
+              migratorInfo.pathSource = pathMigrate.get(tp).getKey();
+              migratorInfo.pathSink = pathMigrate.get(tp).getValue();
+              result.put(tp, migratorInfo);
+            }
+          }
+        });
+    pathMigrate.forEach(
+        (tp, assignments) -> {
+          Set<Integer> fromBroker;
+          Set<Integer> toBroker;
+          if (argument.fromBrokers.containsAll(argument.toBrokers)) {
+            fromBroker = argument.fromBrokers;
+            toBroker = fromBroker;
+          } else {
+            fromBroker = brokerMigrate.get(tp).getKey();
+            toBroker = brokerMigrate.get(tp).getValue();
+          }
+          if (!result.containsKey(tp)) {
+            MigratorInfo migratorInfo = new MigratorInfo();
+            migratorInfo.brokerSource = fromBroker;
+            migratorInfo.brokerSink = toBroker;
+            migratorInfo.pathSource = assignments.getKey();
+            migratorInfo.pathSink = assignments.getValue();
+            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)) {
+                if (!argument.verify) result.remove(tp);
+              } else {
+                MigratorInfo migratorInfo = new MigratorInfo();
+                migratorInfo.brokerSource = fromBroker;
+                migratorInfo.brokerSink = toBroker;
+                migratorInfo.pathSource = assignments.getKey();
+                migratorInfo.pathSink = newPath;
+                result.put(tp, migratorInfo);
+              }
+            }
+          }
         });
     return result;
   }
 
+  static Map<TopicPartition, MigratorInfo> execute(TopicAdmin admin, Argument args) {
+    checkArgs(admin, args);
+    Argument argument = setArguments(admin, args);
+    var brokerMigrate = checkMigratorBroker(admin, argument);
+    var pathMigrate = checkMigratorPath(admin, argument);
+    if (!argument.verify) {
+      brokerMigrator(brokerMigrate, admin);
+      pathMigrator(pathMigrate, admin, argument.toBrokers.iterator().next());
+    }
+    return getResult(brokerMigrate, pathMigrate, argument, admin);
+  }
+
   public static void main(String[] args) throws IOException {
     var argument = ArgumentUtil.parseArgument(new Argument(), args);
     try (var admin = TopicAdmin.of(argument.props())) {
@@ -77,16 +283,20 @@ public class ReplicaCollie {
                           + ", partition: "
                           + tp.partition()
                           + " before: "
-                          + assignments.getKey()
+                          + assignments.brokerSource
+                          + ","
+                          + assignments.pathSource
                           + " after: "
-                          + assignments.getValue()));
+                          + assignments.brokerSink
+                          + ","
+                          + assignments.pathSink));
     }
   }
 
   static class Argument extends BasicArgumentWithPropFile {
     @Parameter(
         names = {"--topics"},
-        description = "Those topics' partitions will get reassigned. Empty menas all topics",
+        description = "Those topics' partitions will get reassigned. Empty means all topics",
         validateWith = ArgumentUtil.NotEmptyString.class,
         converter = ArgumentUtil.StringSetConverter.class)
     public Set<String> topics = Collections.emptySet();
@@ -106,6 +316,20 @@ public class ReplicaCollie {
         converter = ArgumentUtil.IntegerSetConverter.class)
     Set<Integer> toBrokers = Collections.emptySet();
 
+    @Parameter(
+        names = {"--partitions"},
+        description = "all partitions that will be moved",
+        validateWith = ArgumentUtil.NotEmptyString.class,
+        converter = ArgumentUtil.IntegerSetConverter.class)
+    Set<Integer> partitions = Collections.emptySet();
+
+    @Parameter(
+        names = {"--path"},
+        description = "The partition that will be moved to",
+        validateWith = ArgumentUtil.NotEmptyString.class,
+        converter = ArgumentUtil.StringSetConverter.class)
+    Set<String> path = Collections.emptySet();
+
     @Parameter(
         names = {"--verify"},
         description =
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index 80af7463..f04c609c 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -15,7 +15,7 @@ declare -r EXPORTER_VERSION="0.16.1"
 declare -r EXPORTER_PORT=${EXPORTER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
 declare -r BROKER_PORT=${BROKER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
 declare -r CONTAINER_NAME="broker-$BROKER_PORT"
-declare -r BROKER_JMX_PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r BROKER_JMX_PORT="${BROKER_JMX_PORT:-$(($(($RANDOM % 10000)) + 10000))}"
 declare -r ADMIN_NAME="admin"
 declare -r ADMIN_PASSWORD="admin-secret"
 declare -r USER_NAME="user"
diff --git a/docker/start_kafka_tool.sh b/docker/start_kafka_tool.sh
index b51a116d..65997236 100755
--- a/docker/start_kafka_tool.sh
+++ b/docker/start_kafka_tool.sh
@@ -9,6 +9,10 @@ declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null
 declare -r DOCKERFILE=$DOCKER_FOLDER/kafka_tool.dockerfile
 declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
+declare -r ADDRESS=$([[ "$(which ipconfig)" != "" ]] && ipconfig getifaddr en0 || hostname -i)
+declare -r JMX_PORT=${JMX_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false \
+                     -Dcom.sun.management.jmxremote.port=$JMX_PORT -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT -Djava.rmi.server.hostname=$ADDRESS"
 
 # ===================================[functions]===================================
 
@@ -72,9 +76,11 @@ function buildImageIfNeed() {
 
 function runContainer() {
   local args=$1
+  echo "JMX address: $ADDRESS:$JMX_PORT"
   docker run --rm --init \
+    -p $JMX_PORT:$JMX_PORT \
     $IMAGE_NAME \
-    /bin/bash -c "java -jar /tmp/app.jar $args"
+    /bin/bash -c "java $JMX_OPTS -jar /tmp/app.jar $args"
 }
 
 # ===================================[main]===================================


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java
deleted file mode 100644
index 1e8ac86d..00000000
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java
+++ /dev/null
@@ -1,104 +0,0 @@
-package org.astraea.partitioner.nodeLoadMetric;
-
-import java.time.Duration;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.astraea.consumer.Consumer;
-import org.astraea.consumer.Deserializer;
-import org.astraea.consumer.Header;
-import org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner;
-import org.astraea.producer.Producer;
-import org.astraea.service.RequireBrokerCluster;
-import org.astraea.topic.TopicAdmin;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-public class DependencyPartitionTest extends RequireBrokerCluster {
-  private final String brokerList = bootstrapServers();
-  TopicAdmin admin = TopicAdmin.of(bootstrapServers());
-  private final String topicName = "address";
-
-  private Properties initProConfig() {
-    Properties props = new Properties();
-    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
-    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-    props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
-    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
-    props.put("jmx_servers", jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort());
-    return props;
-  }
-
-  @Test
-  void testDependencyPartitioner() {
-    admin.creator().topic(topicName).numberOfPartitions(10).create();
-    var ps = initProConfig();
-    var timestamp = System.currentTimeMillis() + 10;
-    var header = Header.of("a", "b".getBytes());
-    ps.put("producerID", 2);
-    var props =
-        ps.entrySet().stream()
-            .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
-    try (var producer = Producer.builder().configs(props).build()) {
-      var dependencyClient = SmoothWeightPartitioner.dependencyClient(producer.kafkaProducer());
-      dependencyClient.startDependency();
-      var targetPartition = 0;
-      var i = 0;
-      while (i < 20) {
-        var metadata =
-            producer
-                .sender()
-                .topic(topicName)
-                .timestamp(timestamp)
-                .headers(List.of(header))
-                .run()
-                .toCompletableFuture()
-                .get();
-        if (i > 0) Assertions.assertEquals(targetPartition, metadata.partition());
-        targetPartition = metadata.partition();
-        Assertions.assertEquals(topicName, metadata.topic());
-        Assertions.assertEquals(timestamp, metadata.timestamp());
-        i++;
-      }
-      dependencyClient.finishDependency();
-    } catch (ExecutionException | InterruptedException e) {
-      e.printStackTrace();
-    }
-    try (var consumer =
-        Consumer.builder()
-            .brokers(bootstrapServers())
-            .fromBeginning()
-            .topics(Set.of(topicName))
-            .keyDeserializer(Deserializer.STRING)
-            .build()) {
-      var records = consumer.poll(Duration.ofSeconds(10));
-      Assertions.assertEquals(20, records.size());
-      var record = records.iterator().next();
-      Assertions.assertEquals(topicName, record.topic());
-      Assertions.assertEquals(1, record.headers().size());
-      var actualHeader = record.headers().iterator().next();
-      Assertions.assertEquals(header.key(), actualHeader.key());
-      Assertions.assertArrayEquals(header.value(), actualHeader.value());
-    }
-  }
-
-  @Test
-  void testDependencyStateChange() {
-    var props =
-        initProConfig().entrySet().stream()
-            .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
-    var producer1 = Producer.builder().configs(props).build();
-    var producer2 = Producer.builder().configs(props).build();
-    var client1 = SmoothWeightPartitioner.dependencyClient(producer1.kafkaProducer());
-    var client2 = SmoothWeightPartitioner.dependencyClient(producer2.kafkaProducer());
-    client1.startDependency();
-    Assertions.assertEquals(client1.state(), "START_DEPENDENCY");
-    Assertions.assertEquals(client2.state(), "UNINITIALIZED");
-  }
-}
diff --git a/app/src/test/java/org/astraea/performance/DistributionTest.java b/app/src/test/java/org/astraea/performance/DistributionTest.java
index 2cb38401..6ff1e06f 100644
--- a/app/src/test/java/org/astraea/performance/DistributionTest.java
+++ b/app/src/test/java/org/astraea/performance/DistributionTest.java
@@ -1,9 +1,23 @@
 package org.astraea.performance;
 
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class DistributionTest {
+
+  @Test
+  void testFixed() {
+    var distribution = Distribution.fixed();
+    Assertions.assertEquals(
+        1,
+        IntStream.range(0, 10)
+            .mapToObj(ignored -> distribution.get())
+            .collect(Collectors.toSet())
+            .size());
+  }
+
   @Test
   void testLatest() throws InterruptedException {
     var distribution = Distribution.latest();
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 57f72668..a6ef7934 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -60,18 +60,70 @@ public class ManagerTest {
 
     argument.distribution = Distribution.uniform();
     var manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isPresent());
+    Assertions.assertTrue(manager.getKey().length > 0);
 
     argument.distribution = Distribution.zipfian(10);
     manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isPresent());
+    Assertions.assertTrue(manager.getKey().length > 0);
 
     argument.distribution = Distribution.latest();
     manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isPresent());
+    Assertions.assertTrue(manager.getKey().length > 0);
 
-    argument.distribution = null;
+    argument.distribution = Distribution.fixed();
     manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().isEmpty());
+    Assertions.assertTrue(manager.getKey().length > 0);
+  }
+
+  @Test
+  void testConsumerDone() {
+    var argument = new Performance.Argument();
+    var producerMetrics = List.of(new Metrics());
+    var consumerMetrics = List.of(new Metrics());
+
+    argument.exeTime = ExeTime.of("1records");
+    var manager = new Manager(argument, producerMetrics, consumerMetrics);
+    Assertions.assertFalse(manager.consumedDone());
+
+    // Produce one record
+    producerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.consumedDone());
+
+    // Consume one record
+    consumerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.consumedDone());
+
+    manager.producerClosed();
+    Assertions.assertTrue(manager.consumedDone());
+
+    // Test zero consumer. (run for one record)
+    producerMetrics = List.of(new Metrics());
+    consumerMetrics = List.of();
+    manager = new Manager(argument, producerMetrics, consumerMetrics);
+    Assertions.assertFalse(manager.consumedDone());
+
+    // Produce one record
+    producerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.consumedDone());
+
+    manager.producerClosed();
+    Assertions.assertTrue(manager.consumedDone());
+  }
+
+  @Test
+  void testProducerDone() {
+    var argument = new Performance.Argument();
+    argument.exeTime = ExeTime.of("1records");
+    var producerMetrics = List.of(new Metrics());
+    var consumerMetrics = List.of(new Metrics());
+    var manager = new Manager(new Performance.Argument(), producerMetrics, consumerMetrics);
+
+    Assertions.assertFalse(manager.producedDone());
+
+    producerMetrics.get(0).accept(0L, 0L);
+    Assertions.assertFalse(manager.producedDone());
+
+    manager.producerClosed();
+    Assertions.assertTrue(manager.producedDone());
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/MetricsTest.java b/app/src/test/java/org/astraea/performance/MetricsTest.java
index ac9799da..e2c23a78 100644
--- a/app/src/test/java/org/astraea/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/performance/MetricsTest.java
@@ -32,4 +32,15 @@ public class MetricsTest {
     metrics.accept(0L, 1000L);
     Assertions.assertEquals(1000, metrics.bytes());
   }
+
+  @Test
+  void testCurrentBytes() {
+    var metrics = new Metrics();
+
+    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
+    metrics.accept(0L, 100L);
+    metrics.accept(0L, 101L);
+    Assertions.assertEquals(201, metrics.clearAndGetCurrentBytes());
+    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
+  }
 }
diff --git a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
index 14008443..30fe9fed 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
@@ -3,7 +3,10 @@ package org.astraea.topic;
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.io.IOException;
+import java.util.Comparator;
+import java.util.Map;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.TopicPartition;
@@ -27,6 +30,93 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
     test(false);
   }
 
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void testBrokerMigrator() throws IOException, InterruptedException {
+    var topicName = "ReplicaCollieTest";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin
+          .creator()
+          .topic(topicName)
+          .numberOfPartitions(1)
+          .numberOfReplicas((short) 2)
+          .create();
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(5);
+      var partitionReplicas = topicAdmin.replicas(Set.of(topicName));
+      Assertions.assertEquals(1, partitionReplicas.size());
+      var brokerSource =
+          partitionReplicas.get(new TopicPartition(topicName, 0)).stream()
+              .map(Replica::broker)
+              .collect(Collectors.toSet());
+      var brokerSink =
+          topicAdmin.brokerIds().stream().filter(b -> !brokerSource.contains(b)).iterator().next();
+      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate =
+          new TreeMap<>(
+              Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
+      brokerMigrate.put(
+          new TopicPartition(topicName, 0), Map.entry(brokerSource, Set.of(brokerSink)));
+      Assertions.assertEquals(
+          topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).size(), 2);
+      ReplicaCollie.brokerMigrator(brokerMigrate, topicAdmin);
+      Utils.waitFor(
+          () ->
+              topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).size()
+                      == 1
+                  && topicAdmin
+                          .replicas(Set.of(topicName))
+                          .get(new TopicPartition(topicName, 0))
+                          .get(0)
+                          .broker()
+                      == brokerSink);
+    }
+  }
+
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void testPathMigrator() throws IOException, InterruptedException {
+    var topicName = "ReplicaCollieTest";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin
+          .creator()
+          .topic(topicName)
+          .numberOfPartitions(1)
+          .numberOfReplicas((short) 2)
+          .create();
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(5);
+      var partitionReplicas = topicAdmin.replicas(Set.of(topicName));
+      Assertions.assertEquals(1, partitionReplicas.size());
+      var brokerSource = partitionReplicas.get(new TopicPartition(topicName, 0)).get(0).broker();
+      var pathSource =
+          partitionReplicas.get(new TopicPartition(topicName, 0)).stream()
+              .map(Replica::path)
+              .collect(Collectors.toSet());
+      var pathSink =
+          topicAdmin.brokerFolders(Set.of(brokerSource)).get(brokerSource).stream()
+              .filter(p -> !pathSource.contains(p))
+              .iterator()
+              .next();
+      TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate =
+          new TreeMap<>(
+              Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
+      pathMigrate.put(new TopicPartition(topicName, 0), Map.entry(pathSource, Set.of(pathSink)));
+      Assertions.assertFalse(pathSource.contains(pathSink));
+      Assertions.assertEquals(
+          topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).size(), 2);
+      ReplicaCollie.pathMigrator(pathMigrate, topicAdmin, brokerSource);
+      Utils.waitFor(
+          () ->
+              topicAdmin
+                  .replicas(Set.of(topicName))
+                  .get(new TopicPartition(topicName, 0))
+                  .get(0)
+                  .path()
+                  .equals(pathSink));
+      topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0));
+    }
+  }
+
   private void test(boolean verify) throws IOException, InterruptedException {
     var topicName = "ReplicaCollieTest-" + verify;
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
@@ -45,18 +135,22 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
               .filter(Replica::isCurrent)
               .collect(Collectors.toList());
       Assertions.assertEquals(1, replicas.size());
-      var badBroker = replicas.get(0).broker();
+      var badBroker = replicas.stream().map(Replica::broker).collect(Collectors.toSet());
+      var targetBroker =
+          topicAdmin.brokerIds().stream()
+              .filter(b -> !badBroker.contains(b))
+              .collect(Collectors.toSet());
       var argument = new ReplicaCollie.Argument();
-      argument.fromBrokers = Set.of(badBroker);
-      argument.toBrokers = Set.of();
+      argument.fromBrokers = badBroker;
+      argument.toBrokers = targetBroker;
       argument.brokers = bootstrapServers();
+      argument.topics = Set.of(topicName);
+      argument.partitions = Set.of(0);
       argument.verify = verify;
       var result = ReplicaCollie.execute(topicAdmin, argument);
       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));
+      Assertions.assertEquals(badBroker, assignment.brokerSource);
+      Assertions.assertNotEquals(badBroker, assignment.brokerSink);
       if (verify) {
         var currentReplicas = topicAdmin.replicas(Set.of(topicName));
         Assertions.assertEquals(partitionReplicas.size(), currentReplicas.size());
@@ -67,11 +161,17 @@ 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);
+
       } 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);
+              return rs.size() == 2
+                  && rs.stream()
+                      .map(Replica::broker)
+                      .collect(Collectors.toSet())
+                      .containsAll(argument.toBrokers);
             });
       }
     }