Repository: skiptests/astraea
Pull request #461 (2022-07-07 16:57:51)

Introduce HasProducerNodeMetrics and HasProducerTopicMetrics
延續完成如何讓我們的producers可以在某個程度上不仰賴 remote JMX，這隻PR引入兩種producer端的metrics
1. 每個producer會紀錄的broker metrics
2. 每個producer會紀錄的topic metrics


Patch:
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/LeaderElectionTask.java b/app/src/main/java/org/astraea/app/balancer/executor/LeaderElectionTask.java
new file mode 100644
index 00000000..4816567f
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/LeaderElectionTask.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.concurrent.CompletableFuture;
+import org.astraea.app.admin.TopicPartition;
+
+public class LeaderElectionTask {
+
+  private final TopicPartition topicPartition;
+  private final CompletableFuture<Boolean> completableFuture;
+
+  public LeaderElectionTask(RebalanceAdmin admin, TopicPartition topicPartition) {
+    this.topicPartition = topicPartition;
+    this.completableFuture = admin.waitPreferredLeaderSynced(topicPartition);
+  }
+
+  public TopicPartition topicPartition() {
+    return topicPartition;
+  }
+
+  public CompletableFuture<Boolean> completableFuture() {
+    return completableFuture;
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
new file mode 100644
index 00000000..c5cfdbab
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.cost.ClusterInfo;
+
+/**
+ * The wrapper of {@link Admin}. Offer only the essential functionalities & some utilities to
+ * perform rebalance operation.
+ */
+public interface RebalanceAdmin {
+
+  /**
+   * Construct an implementation of {@link RebalanceAdmin}
+   *
+   * @param topicFilter to determine which topics are permitted for balance operation
+   * @param admin the actual {@link Admin} implementation
+   */
+  static RebalanceAdmin of(Admin admin, Predicate<String> topicFilter) {
+    return new RebalanceAdminImpl(topicFilter, admin);
+  }
+
+  /**
+   * Attempt to migrate the target topic/partition to the given replica log placement state. This
+   * method will perform both replica list migration and data directory migration. This method will
+   * return after triggering the migration. It won't wait until the migration processes are
+   * fulfilled.
+   *
+   * @param topicPartition the topic/partition to perform migration
+   * @param expectedPlacement the expected placement after this request accomplished
+   * @return a list of task trackers regarding each log
+   */
+  List<ReplicaMigrationTask> alterReplicaPlacements(
+      TopicPartition topicPartition, List<LogPlacement> expectedPlacement);
+
+  /** @return a {@link CompletableFuture} that indicate the specific log has become synced. */
+  CompletableFuture<Boolean> waitLogSynced(TopicPartitionReplica log, Duration timeout);
+
+  /** @return a {@link CompletableFuture} that indicate the specific log has become synced. */
+  default CompletableFuture<Boolean> waitLogSynced(TopicPartitionReplica log) {
+    return waitLogSynced(log, ChronoUnit.FOREVER.getDuration());
+  }
+
+  /**
+   * @return a {@link CompletableFuture} that indicate the specific topic/partition has its
+   *     preferred leader becomes the actual leader.
+   */
+  CompletableFuture<Boolean> waitPreferredLeaderSynced(
+      TopicPartition topicPartition, Duration timeout);
+
+  /**
+   * @return a {@link CompletableFuture} that indicate the specific topic/partition has its
+   *     preferred leader becomes the actual leader.
+   */
+  default CompletableFuture<Boolean> waitPreferredLeaderSynced(TopicPartition topicPartition) {
+    return waitPreferredLeaderSynced(topicPartition, ChronoUnit.FOREVER.getDuration());
+  }
+
+  /**
+   * Perform preferred leader election for specific topic/partition.
+   *
+   * @param topicPartition the topic/partition to trigger preferred leader election
+   * @return a task tracker to track the election progress.
+   */
+  LeaderElectionTask leaderElection(TopicPartition topicPartition);
+
+  ClusterInfo clusterInfo();
+
+  /**
+   * @return a {@link Predicate<String>} indicate which topic name is allowed to operate by this
+   *     {@link RebalanceAdmin}.
+   */
+  Predicate<String> topicFilter();
+
+  // TODO: add method to apply reassignment bandwidth throttle.
+  // TODO: add method to fetch topic configuration
+  // TODO: add method to fetch broker configuration
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
new file mode 100644
index 00000000..891bb3e6
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
@@ -0,0 +1,259 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.Utils;
+import org.astraea.app.cost.ClusterInfo;
+
+class RebalanceAdminImpl implements RebalanceAdmin {
+
+  private final Predicate<String> topicFilter;
+  private final Admin admin;
+
+  /**
+   * Construct an implementation of {@link RebalanceAdmin}
+   *
+   * @param topicFilter to determine which topics are permitted for balance operation
+   * @param admin the actual {@link Admin} implementation
+   */
+  public RebalanceAdminImpl(Predicate<String> topicFilter, Admin admin) {
+    this.topicFilter = topicFilter;
+    this.admin = admin;
+  }
+
+  private void ensureTopicPermitted(String topic) {
+    if (!topicFilter.test(topic))
+      throw new IllegalArgumentException("Operation to topic \"" + topic + "\" is not permitted");
+  }
+
+  private List<LogPlacement> fetchCurrentPlacement(TopicPartition topicPartition) {
+    ensureTopicPermitted(topicPartition.topic());
+    return admin.replicas(Set.of(topicPartition.topic())).get(topicPartition).stream()
+        .map(replica -> LogPlacement.of(replica.broker(), replica.path()))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  /**
+   * Declare the preferred data directory at certain brokers.
+   *
+   * <p>By default, upon a new log creation with JBOD enabled broker. Kafka broker will pick up a
+   * data directory that has the fewest log maintained to be the data directory for the new log.
+   * This method declares the preferred data directory for a specific topic/partition on the certain
+   * broker. Upon the new log creation for the specific topic/partition on the certain broker. The
+   * preferred data directory will be used as the data directory for the new log, which replaces the
+   * default approach. This gives you the control to decide which data directory the replica log you
+   * are about to migrate will be.
+   *
+   * @param topicPartition the topic/partition to declare preferred data directory
+   * @param preferredPlacements the replica placements with their desired data directory at certain
+   *     brokers
+   */
+  private void declarePreferredDataDirectories(
+      TopicPartition topicPartition, List<LogPlacement> preferredPlacements) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    final var currentPlacement = fetchCurrentPlacement(topicPartition);
+
+    final var currentBrokerAllocation =
+        currentPlacement.stream().map(LogPlacement::broker).collect(Collectors.toUnmodifiableSet());
+
+    // this operation is not supposed to trigger a log movement. But there might be a small window
+    // of time to actually trigger it (race condition).
+    final var declareMap =
+        preferredPlacements.stream()
+            .filter(futurePlacement -> !currentBrokerAllocation.contains(futurePlacement.broker()))
+            .filter(futurePlacement -> futurePlacement.logDirectory().isPresent())
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    LogPlacement::broker, x -> x.logDirectory().orElseThrow()));
+
+    admin
+        .migrator()
+        .partition(topicPartition.topic(), topicPartition.partition())
+        .declarePreferredDir(declareMap);
+  }
+
+  @Override
+  public List<ReplicaMigrationTask> alterReplicaPlacements(
+      TopicPartition topicPartition, List<LogPlacement> expectedPlacement) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    // ensure replica will be placed in the correct data directory at destination broker.
+    declarePreferredDataDirectories(topicPartition, expectedPlacement);
+
+    var currentReplicaBrokers =
+        fetchCurrentPlacement(topicPartition).stream()
+            .map(LogPlacement::broker)
+            .collect(Collectors.toUnmodifiableSet());
+
+    // do cross broker migration
+    admin
+        .migrator()
+        .partition(topicPartition.topic(), topicPartition.partition())
+        .moveTo(
+            expectedPlacement.stream()
+                .map(LogPlacement::broker)
+                .collect(Collectors.toUnmodifiableList()));
+    // do inter-data-directories migration
+    var forCrossDirMigration =
+        expectedPlacement.stream()
+            .filter(placement -> currentReplicaBrokers.contains(placement.broker()))
+            .filter(placement -> placement.logDirectory().isPresent())
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    LogPlacement::broker, placement -> placement.logDirectory().orElseThrow()));
+    admin
+        .migrator()
+        .partition(topicPartition.topic(), topicPartition.partition())
+        .moveTo(forCrossDirMigration);
+
+    return expectedPlacement.stream()
+        .map(
+            log ->
+                new TopicPartitionReplica(
+                    topicPartition.topic(), topicPartition.partition(), log.broker()))
+        .map(log -> new ReplicaMigrationTask(this, log))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  private long getEndTime(Duration timeout) {
+    try {
+      return Math.addExact(System.currentTimeMillis(), timeout.toMillis());
+    } catch (ArithmeticException e) {
+      return Long.MAX_VALUE;
+    }
+  }
+
+  @Override
+  public CompletableFuture<Boolean> waitLogSynced(TopicPartitionReplica log, Duration timeout) {
+    ensureTopicPermitted(log.topic());
+
+    return CompletableFuture.supplyAsync(
+        () -> {
+          // due to the state consistency issue in Kafka broker design. the cluster state returned
+          // from the API might bounce between the `old state` and the `new state` during the very
+          // beginning and accomplishment of the cluster state alteration API. to fix this we use
+          // debounce technique, to ensure the target condition is held over a few successive tries,
+          // which mean the cluster state alteration is considered stable.
+          var debounce = 2;
+          var endTime = getEndTime(timeout);
+          while (!Thread.currentThread().isInterrupted()) {
+            boolean synced =
+                admin.replicas(Set.of(log.topic())).entrySet().stream()
+                    .filter(x -> x.getKey().partition() == log.partition())
+                    .filter(x -> x.getKey().topic().equals(log.topic()))
+                    .flatMap(x -> x.getValue().stream())
+                    .filter(x -> x.broker() == log.brokerId())
+                    .findFirst()
+                    .map(x -> x.inSync() && !x.isFuture())
+                    .orElse(false);
+            // debounce & retrial interval
+            Utils.packException(() -> TimeUnit.MILLISECONDS.sleep(retrialTime.get().toMillis()));
+            debounce = synced ? (debounce - 1) : 2;
+            // synced
+            if (synced && debounce <= 0) return true;
+            // timeout
+            if (System.currentTimeMillis() > endTime) return false;
+          }
+          return false;
+        });
+  }
+
+  @Override
+  public CompletableFuture<Boolean> waitPreferredLeaderSynced(
+      TopicPartition topicPartition, Duration timeout) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    return CompletableFuture.supplyAsync(
+        () -> {
+          // due to the state consistency issue in Kafka broker design. the cluster state returned
+          // from the API might bounce between the `old state` and the `new state` during the very
+          // beginning and accomplishment of the cluster state alteration API. to fix this we use
+          // debounce technique, to ensure the target condition is held over a few successive tries,
+          // which mean the cluster state alteration is considered stable.
+          var debounce = 2;
+          var endTime = getEndTime(timeout);
+          while (!Thread.currentThread().isInterrupted()) {
+            var synced =
+                admin.replicas(Set.of(topicPartition.topic())).entrySet().stream()
+                    .filter(x -> x.getKey().equals(topicPartition))
+                    .findFirst()
+                    .map(Map.Entry::getValue)
+                    .map(
+                        replicas -> {
+                          var preferred =
+                              replicas.stream()
+                                  .filter(Replica::isPreferredLeader)
+                                  .findFirst()
+                                  .orElseThrow();
+                          return preferred.leader();
+                        })
+                    .orElseThrow();
+            // debounce & retrial interval
+            Utils.sleep(retrialTime.get());
+            debounce = synced ? (debounce - 1) : 2;
+            // synced
+            if (synced && debounce <= 0) return true;
+            // timeout
+            if (System.currentTimeMillis() > endTime) return false;
+          }
+          return false;
+        });
+  }
+
+  @Override
+  public LeaderElectionTask leaderElection(TopicPartition topicPartition) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    admin.preferredLeaderElection(topicPartition);
+
+    return new LeaderElectionTask(this, topicPartition);
+  }
+
+  @Override
+  public ClusterInfo clusterInfo() {
+    return admin.clusterInfo(
+        admin.topicNames().stream().filter(topicFilter).collect(Collectors.toUnmodifiableSet()));
+  }
+
+  @Override
+  public Predicate<String> topicFilter() {
+    return topicFilter;
+  }
+
+  private static final AtomicReference<Duration> retrialTime =
+      new AtomicReference<>(Duration.ofSeconds(1));
+
+  // visible for test
+  static void changeRetrialTime(Duration newDebounceTime) {
+    retrialTime.set(newDebounceTime);
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalancePlanExecutor.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalancePlanExecutor.java
new file mode 100644
index 00000000..ba064e58
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalancePlanExecutor.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+
+/** This class associate with the logic of fulfill given rebalance plan. */
+public interface RebalancePlanExecutor {
+
+  /** This method responsible for fulfill a rebalance plan. */
+  void run(RebalanceAdmin rebalanceAdmin, ClusterLogAllocation targetAllocation);
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/ReplicaMigrationTask.java b/app/src/main/java/org/astraea/app/balancer/executor/ReplicaMigrationTask.java
new file mode 100644
index 00000000..47488594
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/ReplicaMigrationTask.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.concurrent.CompletableFuture;
+import org.astraea.app.admin.TopicPartitionReplica;
+
+public class ReplicaMigrationTask {
+
+  private final TopicPartitionReplica log;
+  private final CompletableFuture<Boolean> completableFuture;
+
+  public ReplicaMigrationTask(RebalanceAdmin admin, TopicPartitionReplica log) {
+    this.log = log;
+    this.completableFuture = admin.waitLogSynced(log);
+  }
+
+  public TopicPartitionReplica log() {
+    return log;
+  }
+
+  public CompletableFuture<Boolean> completableFuture() {
+    return completableFuture;
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java b/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
new file mode 100644
index 00000000..e11ff360
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+
+/** Execute every possible migration immediately. */
+public class StraightPlanExecutor implements RebalancePlanExecutor {
+
+  public StraightPlanExecutor() {}
+
+  @Override
+  public void run(RebalanceAdmin rebalanceAdmin, ClusterLogAllocation logAllocation) {
+    final var clusterInfo = rebalanceAdmin.clusterInfo();
+    final var currentLogAllocation = LayeredClusterLogAllocation.of(clusterInfo);
+    final var migrationTargets =
+        ClusterLogAllocation.findNonFulfilledAllocation(currentLogAllocation, logAllocation);
+
+    var executeReplicaMigration =
+        (Function<TopicPartition, List<ReplicaMigrationTask>>)
+            (topicPartition) ->
+                rebalanceAdmin.alterReplicaPlacements(
+                    topicPartition, logAllocation.logPlacements(topicPartition));
+
+    // do log migration
+    migrationTargets.stream()
+        .map(executeReplicaMigration)
+        .flatMap(Collection::stream)
+        .map(ReplicaMigrationTask::completableFuture)
+        .collect(Collectors.toUnmodifiableSet())
+        .forEach(CompletableFuture::join);
+
+    // do leader election
+    migrationTargets.stream()
+        .map(rebalanceAdmin::leaderElection)
+        .map(LeaderElectionTask::completableFuture)
+        .collect(Collectors.toUnmodifiableSet())
+        .forEach(CompletableFuture::join);
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index 0daff1eb..850bc703 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -17,6 +17,8 @@
 package org.astraea.app.balancer.log;
 
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.astraea.app.admin.TopicPartition;
 
@@ -62,5 +64,49 @@ public interface ClusterLogAllocation {
   /** Retrieve the stream of all topic/partition pairs in allocation. */
   Stream<TopicPartition> topicPartitionStream();
 
-  // TODO: add a method to calculate the difference between two ClusterLogAllocation
+  /**
+   * Find a subset of topic/partitions in the source allocation, that has any non-fulfilled log
+   * placement in the given target allocation. Note that the given two allocations must have the
+   * exactly same topic/partitions set. Otherwise, an {@link IllegalArgumentException} will be
+   * raised.
+   */
+  static Set<TopicPartition> findNonFulfilledAllocation(
+      ClusterLogAllocation source, ClusterLogAllocation target) {
+
+    final var sourceTopicPartition =
+        source.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+    final var targetTopicPartition =
+        target.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+
+    if (!sourceTopicPartition.equals(targetTopicPartition))
+      throw new IllegalArgumentException(
+          "source allocation and target allocation has different topic/partition set");
+
+    return sourceTopicPartition.stream()
+        .filter(tp -> !LogPlacement.isMatch(source.logPlacements(tp), target.logPlacements(tp)))
+        .collect(Collectors.toUnmodifiableSet());
+  }
+
+  static String toString(ClusterLogAllocation allocation) {
+    StringBuilder stringBuilder = new StringBuilder();
+
+    allocation
+        .topicPartitionStream()
+        .sorted()
+        .forEach(
+            tp -> {
+              stringBuilder.append("[").append(tp).append("] ");
+
+              allocation
+                  .logPlacements(tp)
+                  .forEach(
+                      log ->
+                          stringBuilder.append(
+                              String.format("%s(%s) ", log.broker(), log.logDirectory())));
+
+              stringBuilder.append(System.lineSeparator());
+            });
+
+    return stringBuilder.toString();
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java b/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
index a0a573c9..9425707f 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
@@ -16,7 +16,9 @@
  */
 package org.astraea.app.balancer.log;
 
+import java.util.List;
 import java.util.Optional;
+import java.util.stream.IntStream;
 
 /** This class describe the placement state of one kafka log. */
 public interface LogPlacement {
@@ -25,6 +27,44 @@ public interface LogPlacement {
 
   Optional<String> logDirectory();
 
+  static boolean isMatch(List<LogPlacement> sourcePlacements, List<LogPlacement> targetPlacements) {
+    if (sourcePlacements.size() != targetPlacements.size()) return false;
+    if (sourcePlacements.equals(targetPlacements)) return true;
+
+    final boolean brokerListMatch =
+        IntStream.range(0, sourcePlacements.size())
+            .allMatch(
+                index ->
+                    sourcePlacements.get(index).broker() == targetPlacements.get(index).broker());
+    if (!brokerListMatch) return false;
+
+    final boolean logDirectoryMatch =
+        IntStream.range(0, sourcePlacements.size())
+            .allMatch(
+                index ->
+                    meetLogDirectoryMigrationRequirement(
+                        sourcePlacements.get(index).logDirectory(),
+                        targetPlacements.get(index).logDirectory()));
+    //noinspection RedundantIfStatement
+    if (!logDirectoryMatch) return false;
+
+    return true;
+  }
+
+  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+  private static boolean meetLogDirectoryMigrationRequirement(
+      Optional<String> sourceDir, Optional<String> targetDir) {
+    // don't care which log directory will eventually be in the destination.
+    if (targetDir.isEmpty()) return true;
+
+    // we care which data directory the target will eventually be, but we don't know.
+    if (sourceDir.isEmpty()) return false;
+
+    // both candidate is specified, if and only if two paths match, will consider as a requirement
+    // meet.
+    return sourceDir.get().equals(targetDir.get());
+  }
+
   static LogPlacement of(int broker) {
     return of(broker, null);
   }
diff --git a/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java b/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java
index 3070f045..bcd761dd 100644
--- a/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/app/metrics/collector/BeanCollector.java
@@ -85,6 +85,7 @@ public class BeanCollector {
 
   public Register register() {
     return new Register() {
+      private boolean local = false;
       private String host;
       private int port = -1;
       private Fetcher fetcher = client -> List.of(KafkaMetrics.Host.jvmMemory(client));
@@ -101,6 +102,14 @@ public class BeanCollector {
         return this;
       }
 
+      @Override
+      public Register local() {
+        this.local = true;
+        this.port = -1;
+        this.host = Utils.hostname();
+        return this;
+      }
+
       @Override
       public Register fetcher(Fetcher fetcher) {
         this.fetcher = Objects.requireNonNull(fetcher);
@@ -153,7 +162,8 @@ public class BeanCollector {
                 if (needUpdate && node.lock.tryLock()) {
                   try {
                     if (node.mBeanClient == null)
-                      node.mBeanClient = clientCreator.apply(host, port);
+                      node.mBeanClient =
+                          local ? MBeanClient.local() : clientCreator.apply(host, port);
                     var beans = fetcher.fetch(node.mBeanClient);
                     // remove old beans if the queue is full
                     for (var t : objects.keySet()) {
diff --git a/app/src/main/java/org/astraea/app/metrics/collector/Register.java b/app/src/main/java/org/astraea/app/metrics/collector/Register.java
index bd8aaa29..d7eb5921 100644
--- a/app/src/main/java/org/astraea/app/metrics/collector/Register.java
+++ b/app/src/main/java/org/astraea/app/metrics/collector/Register.java
@@ -30,6 +30,13 @@ public interface Register {
    */
   Register port(int port);
 
+  /**
+   * declare a receiver based on local metrics client
+   *
+   * @return this register
+   */
+  Register local();
+
   /**
    * @param getters to get metrics from MBeanClient
    * @return this register
diff --git a/app/src/main/java/org/astraea/app/metrics/kafka/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/kafka/KafkaMetrics.java
index 0ae03a4f..090aaf30 100644
--- a/app/src/main/java/org/astraea/app/metrics/kafka/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/kafka/KafkaMetrics.java
@@ -26,6 +26,8 @@ import org.astraea.app.metrics.java.OperatingSystemInfo;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.jmx.BeanQuery;
 import org.astraea.app.metrics.jmx.MBeanClient;
+import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
+import org.astraea.app.metrics.producer.HasProducerTopicMetrics;
 
 public final class KafkaMetrics {
 
@@ -370,4 +372,52 @@ public final class KafkaMetrics {
               BeanQuery.builder().domainName("java.lang").property("type", "Memory").build()));
     }
   }
+
+  public static final class Producer {
+
+    private Producer() {}
+
+    /**
+     * node metrics traced by producer
+     *
+     * @param mBeanClient to query beans
+     * @param brokerId broker ids
+     * @return key is client id used by producer, and value is node metrics traced by each producer
+     */
+    public static Map<String, HasProducerNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.producer")
+                  .property("type", "producer-node-metrics")
+                  .property("node-id", "node-" + brokerId)
+                  .property("client-id", "*")
+                  .build())
+          .stream()
+          .collect(
+              Collectors.toUnmodifiableMap(b -> b.getProperties().get("client-id"), b -> () -> b));
+    }
+
+    /**
+     * topic metrics traced by producer
+     *
+     * @param mBeanClient to query beans
+     * @param topic topic name
+     * @return key is client id used by producer, and value is topic metrics traced by each producer
+     */
+    public static Map<String, HasProducerTopicMetrics> topic(
+        MBeanClient mBeanClient, String topic) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.producer")
+                  .property("type", "producer-topic-metrics")
+                  .property("client-id", "*")
+                  .property("topic", topic)
+                  .build())
+          .stream()
+          .collect(
+              Collectors.toUnmodifiableMap(b -> b.getProperties().get("client-id"), b -> () -> b));
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/producer/HasProducerNodeMetrics.java b/app/src/main/java/org/astraea/app/metrics/producer/HasProducerNodeMetrics.java
new file mode 100644
index 00000000..ee14bb52
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/producer/HasProducerNodeMetrics.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.metrics.producer;
+
+import org.astraea.app.metrics.HasBeanObject;
+
+public interface HasProducerNodeMetrics extends HasBeanObject {
+
+  default double incomingByteRate() {
+    return (double) beanObject().getAttributes().get("incoming-byte-rate");
+  }
+
+  default double incomingByteTotal() {
+    return (double) beanObject().getAttributes().get("incoming-byte-total");
+  }
+
+  default double outgoingByteRate() {
+    return (double) beanObject().getAttributes().get("outgoing-byte-rate");
+  }
+
+  default double outgoingByteTotal() {
+    return (double) beanObject().getAttributes().get("outgoing-byte-total");
+  }
+
+  default double requestLatencyAvg() {
+    return (double) beanObject().getAttributes().get("request-latency-avg");
+  }
+
+  default double requestLatencyMax() {
+    return (double) beanObject().getAttributes().get("request-latency-max");
+  }
+
+  default double requestRate() {
+    return (double) beanObject().getAttributes().get("request-rate");
+  }
+
+  default double requestSizeAvg() {
+    return (double) beanObject().getAttributes().get("request-size-avg");
+  }
+
+  default double requestSizeMax() {
+    return (double) beanObject().getAttributes().get("request-size-max");
+  }
+
+  default double requestTotal() {
+    return (double) beanObject().getAttributes().get("request-total");
+  }
+
+  default double responseRate() {
+    return (double) beanObject().getAttributes().get("response-rate");
+  }
+
+  default double responseTotal() {
+    return (double) beanObject().getAttributes().get("response-total");
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/metrics/producer/HasProducerTopicMetrics.java b/app/src/main/java/org/astraea/app/metrics/producer/HasProducerTopicMetrics.java
new file mode 100644
index 00000000..6f6d4abf
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/producer/HasProducerTopicMetrics.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.metrics.producer;
+
+import org.astraea.app.metrics.HasBeanObject;
+
+public interface HasProducerTopicMetrics extends HasBeanObject {
+
+  default double byteRate() {
+    return (double) beanObject().getAttributes().get("byte-rate");
+  }
+
+  default double byteTotal() {
+    return (double) beanObject().getAttributes().get("byte-total");
+  }
+
+  default double compressionRate() {
+    return (double) beanObject().getAttributes().get("compression-rate");
+  }
+
+  default double recordErrorRate() {
+    return (double) beanObject().getAttributes().get("record-error-rate");
+  }
+
+  default double recordErrorTotal() {
+    return (double) beanObject().getAttributes().get("record-error-total");
+  }
+
+  default double recordRetryRate() {
+    return (double) beanObject().getAttributes().get("record-retry-rate");
+  }
+
+  default double recordRetryTotal() {
+    return (double) beanObject().getAttributes().get("record-retry-total");
+  }
+
+  default double recordSendRate() {
+    return (double) beanObject().getAttributes().get("record-send-rate");
+  }
+
+  default double recordSendTotal() {
+    return (double) beanObject().getAttributes().get("record-send-total");
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
new file mode 100644
index 00000000..95ed015d
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.DataSize;
+import org.astraea.app.common.DataUnit;
+import org.astraea.app.common.Utils;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.Test;
+
+class RebalanceAdminTest extends RequireBrokerCluster {
+
+  @Test
+  void alterReplicaPlacementByList() throws InterruptedException {
+    // arrange
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 1);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(50));
+
+      // scale the replica size from 1 to 3, to the following data dir
+      var logFolder0 = randomElement(logFolders().get(0));
+      var logFolder1 = randomElement(logFolders().get(1));
+      var logFolder2 = randomElement(logFolders().get(2));
+
+      // act
+      var tasks =
+          rebalanceAdmin.alterReplicaPlacements(
+              new TopicPartition(topic, 0),
+              List.of(
+                  LogPlacement.of(0, logFolder0),
+                  LogPlacement.of(1, logFolder1),
+                  LogPlacement.of(2, logFolder2)));
+      tasks.forEach(
+          task -> Utils.packException(() -> task.completableFuture().get(5, TimeUnit.SECONDS)));
+
+      // assert
+      var topicPartition = new TopicPartition(topic, 0);
+      var replicas = admin.replicas(Set.of(topic)).get(topicPartition);
+
+      Assertions.assertEquals(
+          List.of(0, 1, 2), replicas.stream().map(Replica::broker).collect(Collectors.toList()));
+      Assertions.assertEquals(logFolder0, replicas.get(0).path());
+      Assertions.assertEquals(logFolder1, replicas.get(1).path());
+      Assertions.assertEquals(logFolder2, replicas.get(2).path());
+    }
+  }
+
+  // repeat the test so it has higher chance to fail
+  @RepeatedTest(value = 3)
+  void alterReplicaPlacementByDirectory() throws InterruptedException {
+    // arrange
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 1);
+      var topicPartition = new TopicPartition(topic, 0);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      // decrease the debouncing time so the test has higher chance to fail
+      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(150));
+      prepareData(topic, 0, DataUnit.MiB.of(256));
+      Supplier<Replica> replicaNow = () -> admin.replicas(Set.of(topic)).get(topicPartition).get(0);
+      var originalReplica = replicaNow.get();
+      var nextDir =
+          logFolders().get(originalReplica.broker()).stream()
+              .filter(name -> !name.equals(originalReplica.path()))
+              .findAny()
+              .orElseThrow();
+      var expectedPlacement = LogPlacement.of(originalReplica.broker(), nextDir);
+
+      // act, change the dir of the only replica
+      var task =
+          rebalanceAdmin.alterReplicaPlacements(topicPartition, List.of(expectedPlacement)).get(0);
+
+      // assert
+      task.completableFuture().join();
+      var finalReplica = replicaNow.get();
+      Assertions.assertTrue(finalReplica.inSync());
+      Assertions.assertFalse(finalReplica.isFuture());
+      Assertions.assertEquals(originalReplica.broker(), finalReplica.broker());
+    }
+  }
+
+  @Test
+  void clusterInfo() throws InterruptedException {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      // test if all topics are covered
+      final var rebalanceAdmin = RebalanceAdmin.of(admin, (ignore) -> true);
+      final var clusterInfo = rebalanceAdmin.clusterInfo();
+      Assertions.assertEquals(admin.topicNames(), clusterInfo.topics());
+
+      // create topic
+      final var topic = prepareTopic(admin, 3, (short) 1);
+
+      // test if topic filter works
+      final var rebalanceAdmin1 = RebalanceAdmin.of(admin, topic::equals);
+      final var clusterInfo1 = rebalanceAdmin1.clusterInfo();
+      Assertions.assertEquals(Set.of(topic), clusterInfo1.topics());
+    }
+  }
+
+  @Test
+  void checkLogSynced() throws InterruptedException, ExecutionException {
+    // arrange
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 1);
+      var topicPartition = new TopicPartition(topic, 0);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      var beginReplica = admin.replicas().get(topicPartition).get(0);
+      var otherDataDir =
+          admin.brokerFolders().get(beginReplica.broker()).stream()
+              .filter(dir -> !dir.equals(beginReplica.path()))
+              .findAny()
+              .orElseThrow();
+      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(150));
+      prepareData(topic, 0, DataUnit.MiB.of(32));
+      // let two brokers join the replica list
+      admin.migrator().partition(topic, 0).moveTo(List.of(0, 1, 2));
+      // let the existing replica change its directory
+      admin.migrator().partition(topic, 0).moveTo(Map.of(beginReplica.broker(), otherDataDir));
+
+      // act
+      long time0 = System.currentTimeMillis();
+      rebalanceAdmin.waitLogSynced(new TopicPartitionReplica(topic, 0, 0)).get();
+      rebalanceAdmin.waitLogSynced(new TopicPartitionReplica(topic, 0, 1)).get();
+      rebalanceAdmin.waitLogSynced(new TopicPartitionReplica(topic, 0, 2)).get();
+      long time1 = System.currentTimeMillis();
+
+      // assert all replica synced
+      Assertions.assertTrue(
+          admin.replicas(Set.of(topic)).entrySet().stream()
+              .flatMap(x -> x.getValue().stream())
+              .allMatch(Replica::inSync));
+      // assert all data directory migration synced
+      Assertions.assertTrue(
+          admin.replicas(Set.of(topic)).entrySet().stream()
+              .flatMap(x -> x.getValue().stream())
+              .noneMatch(Replica::isFuture));
+      Assertions.assertTrue((time1 - time0) > 100, "This should takes awhile");
+    }
+  }
+
+  @Test
+  void checkPreferredLeaderSynced()
+      throws InterruptedException, ExecutionException, TimeoutException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 3);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      var topicPartition = new TopicPartition(topic, 0);
+
+      var leaderNow =
+          (Supplier<Integer>)
+              () ->
+                  admin.replicas(Set.of(topic)).entrySet().stream()
+                      .filter(x -> x.getKey().topic().equals(topic))
+                      .filter(x -> x.getKey().partition() == 0)
+                      .flatMap(x -> x.getValue().stream())
+                      .filter(Replica::leader)
+                      .findFirst()
+                      .orElseThrow()
+                      .broker();
+
+      int oldLeader = leaderNow.get();
+
+      // change the preferred leader
+      int newPreferredLeader = (oldLeader + 2) % 3;
+      admin
+          .migrator()
+          .partition(topic, 0)
+          .moveTo(List.of(newPreferredLeader, (oldLeader + 1) % 3, oldLeader));
+
+      // assert not leader yet
+      Assertions.assertNotEquals(newPreferredLeader, leaderNow.get());
+
+      // do election
+      rebalanceAdmin.leaderElection(topicPartition).completableFuture().get();
+
+      // wait for this
+      rebalanceAdmin.waitPreferredLeaderSynced(topicPartition).get(5, TimeUnit.SECONDS);
+
+      // assert it is the leader
+      Assertions.assertEquals(newPreferredLeader, leaderNow.get());
+    }
+  }
+
+  @Test
+  void leaderElection() throws InterruptedException, ExecutionException, TimeoutException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 3);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      var topicPartition = new TopicPartition(topic, 0);
+
+      var leaderNow =
+          (Supplier<Integer>)
+              () ->
+                  admin.replicas(Set.of(topic)).entrySet().stream()
+                      .filter(x -> x.getKey().topic().equals(topic))
+                      .filter(x -> x.getKey().partition() == 0)
+                      .flatMap(x -> x.getValue().stream())
+                      .filter(Replica::leader)
+                      .findFirst()
+                      .orElseThrow()
+                      .broker();
+
+      int oldLeader = leaderNow.get();
+
+      // change the preferred leader
+      int newPreferredLeader = (oldLeader + 2) % 3;
+      admin
+          .migrator()
+          .partition(topic, 0)
+          .moveTo(List.of(newPreferredLeader, (oldLeader + 1) % 3, oldLeader));
+
+      // assert not leader yet
+      Assertions.assertNotEquals(newPreferredLeader, leaderNow.get());
+
+      // do election
+      var task = rebalanceAdmin.leaderElection(topicPartition);
+
+      // wait for this
+      rebalanceAdmin.waitPreferredLeaderSynced(topicPartition).get(5, TimeUnit.SECONDS);
+
+      // assert it is the leader now
+      Assertions.assertEquals(newPreferredLeader, leaderNow.get());
+    }
+  }
+
+  @Test
+  void testTopicFilter() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic1 = Utils.randomString();
+      var topic2 = Utils.randomString();
+      var topic3 = Utils.randomString();
+      var topicPartition1 = new TopicPartition(topic1, 0);
+      var topicPartition2 = new TopicPartition(topic2, 0);
+      var topicPartition3 = new TopicPartition(topic3, 0);
+      Stream.of(topic1, topic2, topic3)
+          .forEach(i -> admin.creator().topic(i).numberOfPartitions(1).create());
+      var allowed = List.of(topic1, topic2);
+      Predicate<String> filter = allowed::contains;
+      var rebalanceAdmin = RebalanceAdmin.of(admin, filter);
+
+      Assertions.assertTrue(rebalanceAdmin.topicFilter().test(topic1));
+      Assertions.assertTrue(rebalanceAdmin.topicFilter().test(topic2));
+      Assertions.assertFalse(rebalanceAdmin.topicFilter().test(topic3));
+      Assertions.assertFalse(rebalanceAdmin.topicFilter().test("Something"));
+      Assertions.assertDoesNotThrow(() -> rebalanceAdmin.leaderElection(topicPartition1));
+      Assertions.assertDoesNotThrow(() -> rebalanceAdmin.leaderElection(topicPartition2));
+      Assertions.assertThrows(
+          IllegalArgumentException.class, () -> rebalanceAdmin.leaderElection(topicPartition3));
+    }
+  }
+
+  String prepareTopic(Admin admin, int partition, short replica) throws InterruptedException {
+    var topicName = Utils.randomString();
+
+    admin
+        .creator()
+        .topic(topicName)
+        .numberOfPartitions(partition)
+        .numberOfReplicas(replica)
+        .create();
+    TimeUnit.SECONDS.sleep(1);
+
+    return topicName;
+  }
+
+  RebalanceAdmin prepareRebalanceAdmin(Admin admin) {
+    return RebalanceAdmin.of(admin, (ignore) -> true);
+  }
+
+  void prepareData(String topic, int partition, DataSize dataSize) {
+    var dummy = new byte[1024];
+    int sends = dataSize.measurement(DataUnit.KiB).intValue();
+    try (var producer = Producer.of(bootstrapServers())) {
+      var sender = producer.sender().topic(topic).partition(partition).value(dummy);
+      IntStream.range(0, sends)
+          .parallel()
+          .mapToObj(i -> sender.run().toCompletableFuture())
+          .collect(Collectors.toSet())
+          .forEach(CompletableFuture::join);
+    }
+  }
+
+  <T> T randomElement(Collection<T> collection) {
+    return collection.stream()
+        .skip(ThreadLocalRandom.current().nextInt(0, collection.size()))
+        .findFirst()
+        .orElseThrow();
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
new file mode 100644
index 00000000..59873bfa
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class StraightPlanExecutorTest extends RequireBrokerCluster {
+
+  @Test
+  void testRun() throws InterruptedException {
+    // arrange
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
+      admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 2).create();
+      TimeUnit.SECONDS.sleep(2);
+      final var originalAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      TimeUnit.SECONDS.sleep(3);
+      final var broker0 = 0;
+      final var broker1 = 1;
+      final var logFolder0 = logFolders().get(broker0).stream().findAny().orElseThrow();
+      final var logFolder1 = logFolders().get(broker1).stream().findAny().orElseThrow();
+      final var onlyPlacement =
+          List.of(LogPlacement.of(broker0, logFolder0), LogPlacement.of(broker1, logFolder1));
+      final var allocationMap =
+          IntStream.range(0, 10)
+              .mapToObj(i -> new TopicPartition(topicName, i))
+              .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
+      final var expectedAllocation = LayeredClusterLogAllocation.of(allocationMap);
+      final var expectedTopicPartition =
+          expectedAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
+
+      // act
+      new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
+
+      // assert
+      final var currentAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var currentTopicPartition =
+          currentAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
+      expectedTopicPartition.forEach(
+          topicPartition ->
+              Assertions.assertEquals(
+                  expectedAllocation.logPlacements(topicPartition),
+                  currentAllocation.logPlacements(topicPartition),
+                  "Testing for " + topicPartition));
+
+      System.out.println("Expected:");
+      System.out.println(ClusterLogAllocation.toString(expectedAllocation));
+      System.out.println("Current:");
+      System.out.println(ClusterLogAllocation.toString(currentAllocation));
+      System.out.println("Original:");
+      System.out.println(ClusterLogAllocation.toString(originalAllocation));
+    }
+  }
+
+  @Test
+  void testRunNoDirSpecified() throws InterruptedException {
+    // arrange
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
+      admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 3).create();
+      TimeUnit.SECONDS.sleep(2);
+      final var originalAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      TimeUnit.SECONDS.sleep(3);
+      final var broker0 = 0;
+      final var broker1 = 1;
+      final var broker2 = 2;
+      final var onlyPlacement =
+          List.of(LogPlacement.of(broker0), LogPlacement.of(broker1), LogPlacement.of(broker2));
+      final var allocationMap =
+          IntStream.range(0, 10)
+              .mapToObj(i -> new TopicPartition(topicName, i))
+              .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
+      final var expectedAllocation = LayeredClusterLogAllocation.of(allocationMap);
+      final var expectedTopicPartition =
+          expectedAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
+
+      // act
+      new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
+
+      // assert
+      final var currentAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var currentTopicPartition =
+          currentAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
+      expectedTopicPartition.forEach(
+          topicPartition ->
+              Assertions.assertTrue(
+                  LogPlacement.isMatch(
+                      currentAllocation.logPlacements(topicPartition),
+                      expectedAllocation.logPlacements(topicPartition))));
+
+      System.out.println("Expected:");
+      System.out.println(ClusterLogAllocation.toString(expectedAllocation));
+      System.out.println("Current:");
+      System.out.println(ClusterLogAllocation.toString(currentAllocation));
+      System.out.println("Original:");
+      System.out.println(ClusterLogAllocation.toString(originalAllocation));
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java b/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
index 233f671c..42556139 100644
--- a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
@@ -202,4 +202,49 @@ class LayeredClusterLogAllocationTest {
     Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 8, 1, "somewhere"));
     Assertions.assertDoesNotThrow(() -> extended.letReplicaBecomeLeader(toModify, 0));
   }
+
+  @Test
+  void findNonFulfilledAllocation() {
+    final var clusterInfo = ClusterInfoProvider.fakeClusterInfo(3, 10, 10, 2);
+    final var a = LayeredClusterLogAllocation.of(clusterInfo);
+    final var b = LayeredClusterLogAllocation.of(clusterInfo);
+    Assertions.assertEquals(Set.of(), ClusterLogAllocation.findNonFulfilledAllocation(a, b));
+
+    final var source = LayeredClusterLogAllocation.of(clusterInfo);
+    final var oneTopicPartition = source.topicPartitionStream().findFirst().orElseThrow();
+    final var twoTopicPartition = source.topicPartitionStream().skip(1).findFirst().orElseThrow();
+
+    final var target0 = LayeredClusterLogAllocation.of(a);
+    target0.migrateReplica(oneTopicPartition, 0, 0, "/somewhere");
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target0));
+
+    final var target1 = LayeredClusterLogAllocation.of(a);
+    target1.migrateReplica(oneTopicPartition, 0, 2);
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target1));
+
+    final var target2 = LayeredClusterLogAllocation.of(a);
+    target2.letReplicaBecomeLeader(oneTopicPartition, 1);
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target2));
+
+    final var target3 = LayeredClusterLogAllocation.of(a);
+    target3.migrateReplica(oneTopicPartition, 0, 2);
+    target3.migrateReplica(oneTopicPartition, 2, 2, "/somewhere");
+    target3.letReplicaBecomeLeader(twoTopicPartition, 1);
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition, twoTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target3));
+
+    final var map4 = a.topicPartitionStream().collect(Collectors.toMap(x -> x, a::logPlacements));
+    map4.put(new TopicPartition("NewTopic", 0), List.of(LogPlacement.of(0, "?")));
+    final var target4 = LayeredClusterLogAllocation.of(map4);
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> ClusterLogAllocation.findNonFulfilledAllocation(source, target4));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java b/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
new file mode 100644
index 00000000..6592b5e6
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.log;
+
+import java.util.List;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+class LogPlacementTest {
+
+  @Test
+  @DisplayName("Equals")
+  void isMatch0() {
+    final var placement0 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var placement1 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    Assertions.assertTrue(LogPlacement.isMatch(placement0, placement1));
+  }
+
+  @Test
+  @DisplayName("Wrong order")
+  void noMatch0() {
+    final var placement0 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var placement1 =
+        List.of(LogPlacement.of(1, "/B"), LogPlacement.of(0, "/A"), LogPlacement.of(2, "/C"));
+    Assertions.assertFalse(LogPlacement.isMatch(placement0, placement1));
+  }
+
+  @Test
+  @DisplayName("No log dir name match")
+  void noMatch1() {
+    final var placement0 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var placement1 =
+        List.of(LogPlacement.of(0, "/Aaa"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    Assertions.assertFalse(LogPlacement.isMatch(placement0, placement1));
+  }
+
+  @Test
+  @DisplayName("Optional log dir")
+  void noMatch2() {
+    final var sourcePlacement =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var targetPlacement =
+        List.of(LogPlacement.of(0), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    // don't care which log dir placement[0] will eventually be.
+    Assertions.assertTrue(LogPlacement.isMatch(sourcePlacement, targetPlacement));
+  }
+
+  @Test
+  @DisplayName("Optional log dir")
+  void noMatch3() {
+    final var sourcePlacement =
+        List.of(LogPlacement.of(0), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var targetPlacement =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    // do care which log dir placement[0] will eventually be.
+    Assertions.assertFalse(LogPlacement.isMatch(sourcePlacement, targetPlacement));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java b/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
index 8d517fac..5439a2a5 100644
--- a/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
@@ -25,12 +25,14 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.app.common.Utils;
 import org.astraea.app.concurrent.Executor;
 import org.astraea.app.concurrent.State;
 import org.astraea.app.concurrent.ThreadPool;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.jmx.MBeanClient;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -297,4 +299,24 @@ public class BeanCollectorTest {
       Assertions.assertEquals(obj.createdTimestamp(), objs.iterator().next().createdTimestamp());
     }
   }
+
+  @Test
+  void testLocal() {
+    var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(1))
+            .clientCreator(clientCreator)
+            .build();
+    try (var receiver =
+        collector
+            .register()
+            .local()
+            .fetcher(client -> List.of(KafkaMetrics.Host.jvmMemory(client)))
+            .build()) {
+
+      // wait for updating cache
+      Utils.sleep(Duration.ofSeconds(1));
+      Assertions.assertNotEquals(0, receiver.current().size());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
new file mode 100644
index 00000000..4d4bc0af
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.metrics.producer;
+
+import java.util.concurrent.ExecutionException;
+import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.jmx.MBeanClient;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class HasProducerNodeMetricsTest extends RequireSingleBrokerCluster {
+
+  @Test
+  void testAttributes() throws ExecutionException, InterruptedException {
+    var topic = Utils.randomString(10);
+    try (var producer = Producer.of(bootstrapServers())) {
+      producer.sender().topic(topic).run().toCompletableFuture().get();
+      var metrics = KafkaMetrics.Producer.node(MBeanClient.local(), brokerIds().iterator().next());
+      Assertions.assertEquals(1, metrics.size());
+      var producerNodeMetrics = metrics.get("producer-1");
+      Assertions.assertNotEquals(0D, producerNodeMetrics.incomingByteRate());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.incomingByteTotal());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.outgoingByteRate());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.outgoingByteTotal());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.requestLatencyAvg());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.requestLatencyMax());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.requestRate());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.requestSizeAvg());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.requestSizeMax());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.requestTotal());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.responseRate());
+      Assertions.assertNotEquals(0D, producerNodeMetrics.responseTotal());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java
new file mode 100644
index 00000000..eab881af
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.metrics.producer;
+
+import java.util.concurrent.ExecutionException;
+import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.jmx.MBeanClient;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class HasProducerTopicMetricsTest extends RequireSingleBrokerCluster {
+
+  @Test
+  void testAttributes() throws ExecutionException, InterruptedException {
+    var topic = Utils.randomString(10);
+    try (var producer = Producer.of(bootstrapServers())) {
+      producer.sender().topic(topic).run().toCompletableFuture().get();
+      var metrics = KafkaMetrics.Producer.topic(MBeanClient.local(), topic);
+      Assertions.assertEquals(1, metrics.size());
+      var producerTopicMetrics = metrics.get("producer-1");
+      Assertions.assertNotEquals(0D, producerTopicMetrics.byteRate());
+      Assertions.assertNotEquals(0D, producerTopicMetrics.byteTotal());
+      Assertions.assertEquals(1D, producerTopicMetrics.compressionRate());
+      Assertions.assertEquals(0D, producerTopicMetrics.recordErrorRate());
+      Assertions.assertEquals(0D, producerTopicMetrics.recordErrorTotal());
+      Assertions.assertEquals(0D, producerTopicMetrics.recordRetryRate());
+      Assertions.assertEquals(0D, producerTopicMetrics.recordRetryTotal());
+      Assertions.assertNotEquals(0D, producerTopicMetrics.recordSendRate());
+      Assertions.assertNotEquals(0D, producerTopicMetrics.recordSendTotal());
+    }
+  }
+}