Repository: skiptests/astraea (java)

Issue:
`BalancerHandler` 沒辦法對有多個 replica 的 topic 套用計劃
在 3 個節點的叢集建立 2 個 replica 或 3 個 replica topic，計劃生成可以順利執行，但套用計劃時會被 sanity check 擋下來

```
{
  "code": 400,
  "message": "java.lang.IllegalStateException: The cluster state has been changed significantly. The following topic/partitions have different replica list(lookup the moment of plan generation): [topicA-83, topicB-53, topicA-84, topicB-58, topicA-90, topicA-85, topicB-61, topicB-62, topicA-94, topicA-95, topicB-55, topicB-66, topicB-67, topicB-68, topicA-88, topicB-70, topicB-71, topicB-72, topicB-73, topicB-76, topicB-78, topicB-81, topicB-82, topicB-83, topicB-88, topicB-91, topicB-92, topicB-93, topicB-95, topicB-96, topicB-98, topicA-2, topicA-3, topicA-4, topicA-7, topicA-8, topicA-10, topicA-13, topicA-14, topicA-15, topicA-18, topicA-19, topicA-22, topicA-23, topicA-25, topicA-27, topicA-30, topicB-1, topicB-2, topicA-34, topicB-6, topicB-7, topicB-8, topicA-39, topicA-40, topicB-11, topicA-42, topicB-12, topicA-43, topicB-13, topicA-45, topicB-15, topicA-47, topicA-48, topicB-18, topicA-50, topicB-21, topicB-22, topicA-53, topicB-23, topicB-26, topicB-27, topicA-58, topicB-28, topicA-60, topicB-30, topicB-31, topicA-62, topicA-65, topicB-35, topicB-36, topicA-67, topicA-68, topicB-38, topicA-70, topicB-41, topicB-42, topicA-74, topicA-75, topicB-32, topicA-63, topicB-47, topicA-78, topicA-64, topicA-54, topicB-50, topicB-51, topicA-82, topicB-52]"
}
```

可能需要檢查一下條件有沒有寫錯

Pull request description:
[WEB] Fix the placement output order of `Change#before` and `Change#after`
Resolve #1320

`BalancerHandler.Change` 的 before/after 輸出順序有一定的規則，第一個 element 必須要是 Partition 的 Preferred Leader。這個 PR 新增這個排序規則還有寫測試確保這個行為。

Patch:
diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
index f0d183cc..75ebe40f 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -23,6 +23,7 @@ import java.util.Collection;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -171,19 +172,9 @@ class BalancerHandler implements Handler {
                                       .stream()
                                       .map(
                                           tp ->
-                                              new Change(
-                                                  tp.topic(),
-                                                  tp.partition(),
-                                                  // only log the size from source replicas
-                                                  currentClusterInfo.replicas(tp).stream()
-                                                      .map(
-                                                          r ->
-                                                              new Placement(
-                                                                  r, Optional.of(r.size())))
-                                                      .collect(Collectors.toList()),
-                                                  p.proposal().replicas(tp).stream()
-                                                      .map(r -> new Placement(r, Optional.empty()))
-                                                      .collect(Collectors.toList())))
+                                              Change.from(
+                                                  currentClusterInfo.replicas(tp),
+                                                  p.proposal().replicas(tp)))
                                       .collect(Collectors.toUnmodifiableList()))
                           .orElse(List.of());
                   var report =
@@ -197,7 +188,7 @@ class BalancerHandler implements Handler {
                               .solution()
                               .map(p -> migrationCosts(p.moveCost()))
                               .orElseGet(List::of));
-                  return new PlanInfo(report, bestPlan);
+                  return new PlanInfo(report, currentClusterInfo, bestPlan);
                 })
             .whenComplete(
                 (result, error) -> {
@@ -473,12 +464,16 @@ class BalancerHandler implements Handler {
                           .map(x -> Map.entry(x.nodeInfo().id(), x.path()))
                           .collect(Collectors.toUnmodifiableList());
                   var expectedReplicaList =
-                      Stream.concat(
-                              change.before.stream().limit(1),
-                              change.before.stream()
-                                  .skip(1)
-                                  .sorted(Comparator.comparing(x -> x.brokerId)))
-                          .map(x -> Map.entry(x.brokerId, x.directory))
+                      thePlanInfo
+                          .associatedClusterInfo
+                          .replicas(TopicPartition.of(change.topic, change.partition))
+                          .stream()
+                          // have to compare by isLeader instead of isPreferredLeader.
+                          // since the leadership is what affects the direction of traffic load.
+                          // Any bandwidth related cost function should calculate load by leadership
+                          // instead of preferred leadership
+                          .sorted(Comparator.comparing(Replica::isLeader).reversed())
+                          .map(replica -> Map.entry(replica.nodeInfo().id(), replica.path()))
                           .collect(Collectors.toUnmodifiableList());
                   return !expectedReplicaList.equals(currentReplicaList);
                 })
@@ -539,6 +534,28 @@ class BalancerHandler implements Handler {
     final List<Placement> before;
     final List<Placement> after;
 
+    static Change from(Collection<Replica> before, Collection<Replica> after) {
+      if (before.size() == 0) throw new NoSuchElementException("Empty replica list was given");
+      if (after.size() == 0) throw new NoSuchElementException("Empty replica list was given");
+      var tp = before.stream().findAny().orElseThrow().topicPartition();
+      if (!before.stream().allMatch(r -> r.topicPartition().equals(tp)))
+        throw new IllegalArgumentException("Some replica come from different topic/partition");
+      if (!after.stream().allMatch(r -> r.topicPartition().equals(tp)))
+        throw new IllegalArgumentException("Some replica come from different topic/partition");
+      return new Change(
+          tp.topic(),
+          tp.partition(),
+          // only log the size from source replicas
+          before.stream()
+              .sorted(Comparator.comparing(Replica::isPreferredLeader).reversed())
+              .map(r -> new Placement(r, Optional.of(r.size())))
+              .collect(Collectors.toList()),
+          after.stream()
+              .sorted(Comparator.comparing(Replica::isPreferredLeader).reversed())
+              .map(r -> new Placement(r, Optional.empty()))
+              .collect(Collectors.toList()));
+    }
+
     Change(String topic, int partition, List<Placement> before, List<Placement> after) {
       this.topic = topic;
       this.partition = partition;
@@ -592,10 +609,13 @@ class BalancerHandler implements Handler {
 
   static class PlanInfo {
     private final Report report;
+    private final ClusterInfo<Replica> associatedClusterInfo;
     private final Balancer.Plan associatedPlan;
 
-    PlanInfo(Report report, Balancer.Plan associatedPlan) {
+    PlanInfo(
+        Report report, ClusterInfo<Replica> associatedClusterInfo, Balancer.Plan associatedPlan) {
       this.report = report;
+      this.associatedClusterInfo = associatedClusterInfo;
       this.associatedPlan = associatedPlan;
     }
   }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
index 9647808e..85bb011f 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -23,6 +23,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -31,6 +32,7 @@ import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -39,6 +41,7 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import org.astraea.app.web.BalancerHandler.BalancerPostRequest;
 import org.astraea.app.web.BalancerHandler.CostWeight;
 import org.astraea.common.Configuration;
@@ -47,6 +50,7 @@ import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
 import org.astraea.common.admin.ClusterBean;
 import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.admin.ClusterInfoBuilder;
 import org.astraea.common.admin.NodeInfo;
 import org.astraea.common.admin.Replica;
 import org.astraea.common.admin.TopicPartition;
@@ -175,6 +179,11 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
   }
 
   private static List<String> createAndProduceTopic(int topicCount) {
+    return createAndProduceTopic(topicCount, 3, (short) 1, true);
+  }
+
+  private static List<String> createAndProduceTopic(
+      int topicCount, int partitions, short replicas, boolean skewed) {
     try (var admin = Admin.of(bootstrapServers())) {
       var topics =
           IntStream.range(0, topicCount)
@@ -185,18 +194,22 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
             admin
                 .creator()
                 .topic(topic)
-                .numberOfPartitions(3)
-                .numberOfReplicas((short) 1)
+                .numberOfPartitions(partitions)
+                .numberOfReplicas(replicas)
                 .run()
                 .toCompletableFuture()
                 .join();
-            Utils.sleep(Duration.ofSeconds(1));
-            admin
-                .moveToBrokers(
-                    admin.topicPartitions(Set.of(topic)).toCompletableFuture().join().stream()
-                        .collect(Collectors.toMap(tp -> tp, ignored -> List.of(1))))
-                .toCompletableFuture()
-                .join();
+            if (skewed) {
+              Utils.sleep(Duration.ofSeconds(1));
+              var placement =
+                  brokerIds().stream().limit(replicas).collect(Collectors.toUnmodifiableList());
+              admin
+                  .moveToBrokers(
+                      admin.topicPartitions(Set.of(topic)).toCompletableFuture().join().stream()
+                          .collect(Collectors.toMap(tp -> tp, ignored -> placement)))
+                  .toCompletableFuture()
+                  .join();
+            }
           });
       Utils.sleep(Duration.ofSeconds(3));
       try (var producer = Producer.of(bootstrapServers())) {
@@ -407,7 +420,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
   @Timeout(value = 60)
   void testPut() {
     // arrange
-    createAndProduceTopic(3);
+    createAndProduceTopic(3, 10, (short) 2, false);
     try (var admin = Admin.of(bootstrapServers())) {
       var theExecutor = new NoOpExecutor();
       var handler = new BalancerHandler(admin, theExecutor);
@@ -1072,6 +1085,132 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testChangeOrder() {
+    // arrange
+    var sourcePlacement =
+        IntStream.range(0, 10)
+            .mapToObj(partition -> Map.entry(ThreadLocalRandom.current().nextInt(), partition))
+            .sorted(Map.Entry.comparingByKey())
+            .map(Map.Entry::getValue)
+            .collect(Collectors.toUnmodifiableList());
+    var destPlacement =
+        IntStream.range(0, 10)
+            .mapToObj(partition -> Map.entry(ThreadLocalRandom.current().nextInt(), partition))
+            .sorted(Map.Entry.comparingByKey())
+            .map(Map.Entry::getValue)
+            .collect(Collectors.toUnmodifiableList());
+    var base =
+        ClusterInfoBuilder.builder()
+            .addNode(Set.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))
+            .addFolders(Map.of(0, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(1, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(2, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(3, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(4, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(5, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(6, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(7, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(8, Set.of("/folder0", "/folder1", "/folder2")))
+            .addFolders(Map.of(9, Set.of("/folder0", "/folder1", "/folder2")))
+            .build();
+    var srcIter = sourcePlacement.iterator();
+    var srcPrefIter = Stream.iterate(true, (ignore) -> false).iterator();
+    var srcDirIter = Stream.generate(() -> "/folder0").iterator();
+    var sourceCluster =
+        ClusterInfoBuilder.builder(base)
+            .addTopic(
+                "Pipeline",
+                1,
+                (short) 10,
+                r ->
+                    Replica.builder(r)
+                        .nodeInfo(base.node(srcIter.next()))
+                        .isPreferredLeader(srcPrefIter.next())
+                        .path(srcDirIter.next())
+                        .build())
+            .build();
+    var dstIter = destPlacement.iterator();
+    var dstPrefIter = Stream.iterate(true, (ignore) -> false).iterator();
+    var dstDirIter = Stream.generate(() -> "/folder1").iterator();
+    var destCluster =
+        ClusterInfoBuilder.builder(base)
+            .addTopic(
+                "Pipeline",
+                1,
+                (short) 10,
+                r ->
+                    Replica.builder(r)
+                        .nodeInfo(base.node(dstIter.next()))
+                        .isPreferredLeader(dstPrefIter.next())
+                        .path(dstDirIter.next())
+                        .build())
+            .build();
+
+    // act
+    var change =
+        BalancerHandler.Change.from(
+            sourceCluster.replicas("Pipeline"), destCluster.replicas("Pipeline"));
+
+    // assert
+    Assertions.assertEquals("Pipeline", change.topic);
+    Assertions.assertEquals(0, change.partition);
+    Assertions.assertEquals(
+        sourcePlacement.get(0), change.before.get(0).brokerId, "First replica is preferred leader");
+    Assertions.assertEquals(
+        destPlacement.get(0), change.after.get(0).brokerId, "First replica is preferred leader");
+    Assertions.assertEquals(
+        Set.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9),
+        change.before.stream().map(x -> x.brokerId).collect(Collectors.toUnmodifiableSet()),
+        "No node ignored");
+    Assertions.assertEquals(
+        Set.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9),
+        change.after.stream().map(x -> x.brokerId).collect(Collectors.toUnmodifiableSet()),
+        "No node ignored");
+    Assertions.assertTrue(
+        change.before.stream().map(x -> x.directory).allMatch(x -> x.equals("/folder0")),
+        "Correct folder");
+    Assertions.assertTrue(
+        change.after.stream().map(x -> x.directory).allMatch(x -> x.equals("/folder1")),
+        "Correct folder");
+    Assertions.assertTrue(change.before.stream().allMatch(x -> x.size.isPresent()), "Has size");
+    Assertions.assertTrue(change.after.stream().noneMatch(x -> x.size.isPresent()), "No size");
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            BalancerHandler.Change.from(
+                ClusterInfoBuilder.builder(base)
+                    .addTopic("Pipeline", 1, (short) 3)
+                    .build()
+                    .replicas(),
+                ClusterInfoBuilder.builder(base)
+                    .addTopic("Pipeline", 5, (short) 3)
+                    .build()
+                    .replicas()),
+        "Should be a replica list");
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            BalancerHandler.Change.from(
+                ClusterInfoBuilder.builder(base)
+                    .addTopic("Pipeline", 5, (short) 3)
+                    .build()
+                    .replicas(),
+                ClusterInfoBuilder.builder(base)
+                    .addTopic("Pipeline", 1, (short) 3)
+                    .build()
+                    .replicas()),
+        "Should be a replica list");
+    Assertions.assertThrows(
+        NoSuchElementException.class,
+        () -> BalancerHandler.Change.from(sourceCluster.replicas(), Set.of()));
+    Assertions.assertThrows(
+        NoSuchElementException.class,
+        () -> BalancerHandler.Change.from(Set.of(), sourceCluster.replicas()));
+    Assertions.assertThrows(
+        NoSuchElementException.class, () -> BalancerHandler.Change.from(Set.of(), Set.of()));
+  }
+
   /** Submit the plan and wait until it generated. */
   private BalancerHandler.PlanExecutionProgress submitPlanGeneration(
       BalancerHandler handler, Map<String, Object> requestBody) {