Repository: skiptests/astraea
Pull request #607 (2022-08-21 18:47:21)

Support to execute migration plans by Reassignment Web API
支援一次執行多個migration plans，這會方便我們之後執行balancer


Patch:
diff --git a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
index 380340b6..489c8dc4 100644
--- a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
+++ b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
@@ -24,6 +24,7 @@ import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
 
 public class ReassignmentHandler implements Handler {
+  static final String PLANS_KEY = "plans";
   static final String TOPIC_KEY = "topic";
   static final String PARTITION_KEY = "partition";
   static final String FROM_KEY = "from";
@@ -36,23 +37,31 @@ public class ReassignmentHandler implements Handler {
   }
 
   @Override
-  public Response post(PostRequest request) {
-    // case 0: move replica to another folder
-    if (request.has(BROKER_KEY, TOPIC_KEY, PARTITION_KEY, TO_KEY)) {
-      admin
-          .migrator()
-          .partition(request.value(TOPIC_KEY), request.intValue(PARTITION_KEY))
-          .moveTo(Map.of(request.intValue(BROKER_KEY), request.value(TO_KEY)));
-      return Response.ACCEPT;
-    }
-    // case 1: move replica to another broker
-    if (request.has(TOPIC_KEY, PARTITION_KEY, TO_KEY)) {
-      admin
-          .migrator()
-          .partition(request.value(TOPIC_KEY), request.intValue(PARTITION_KEY))
-          .moveTo(request.intValues(TO_KEY));
-      return Response.ACCEPT;
-    }
+  public Response post(PostRequest requests) {
+    var rs =
+        requests.requests(PLANS_KEY).stream()
+            .map(
+                request -> {
+                  // case 0: move replica to another folder
+                  if (request.has(BROKER_KEY, TOPIC_KEY, PARTITION_KEY, TO_KEY)) {
+                    admin
+                        .migrator()
+                        .partition(request.value(TOPIC_KEY), request.intValue(PARTITION_KEY))
+                        .moveTo(Map.of(request.intValue(BROKER_KEY), request.value(TO_KEY)));
+                    return Response.ACCEPT;
+                  }
+                  // case 1: move replica to another broker
+                  if (request.has(TOPIC_KEY, PARTITION_KEY, TO_KEY)) {
+                    admin
+                        .migrator()
+                        .partition(request.value(TOPIC_KEY), request.intValue(PARTITION_KEY))
+                        .moveTo(request.intValues(TO_KEY));
+                    return Response.ACCEPT;
+                  }
+                  return Response.BAD_REQUEST;
+                })
+            .collect(Collectors.toUnmodifiableList());
+    if (!rs.isEmpty() && rs.stream().allMatch(r -> r == Response.ACCEPT)) return Response.ACCEPT;
     return Response.BAD_REQUEST;
   }
 
diff --git a/docs/web_server/web_api_reassignments_chinese.md b/docs/web_server/web_api_reassignments_chinese.md
index 18bee020..f37e5120 100644
--- a/docs/web_server/web_api_reassignments_chinese.md
+++ b/docs/web_server/web_api_reassignments_chinese.md
@@ -63,11 +63,11 @@ cURL 範例
 ```shell
 curl -X POST http://localhost:8001/reassignments \
     -H "Content-Type: application/json" \
-    -d '{
+    -d '"plans":[{
     "topic": "chia", 
     "partition": 0,
     "to": [1003]
-    }' 
+    }]' 
 ```
 
 ## 變更 replica 的資料路徑
@@ -90,10 +90,10 @@ cURL 範例
 ```shell
 curl -X POST http://localhost:8001/reassignments \
     -H "Content-Type: application/json" \
-    -d '{
+    -d '"plans":[{
     "topic": "chia", 
     "partition": 0,
     "broker": 1003
     "to": "/tmp/data"
-    }' 
+    }]' 
 ```
\ No newline at end of file


Test patch:
diff --git a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
index 880301c7..39e84c69 100644
--- a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
@@ -41,17 +41,18 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
           admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).broker();
       var nextBroker = brokerIds().stream().filter(i -> i != currentBroker).findAny().get();
 
-      Assertions.assertEquals(
-          Response.ACCEPT,
-          handler.post(
-              PostRequest.of(
-                  Map.of(
-                      ReassignmentHandler.TOPIC_KEY,
-                      topicName,
-                      ReassignmentHandler.PARTITION_KEY,
-                      "0",
-                      ReassignmentHandler.TO_KEY,
-                      "[\"" + nextBroker + "\"]"))));
+      var body =
+          String.format(
+              "{\"%s\": [{\"%s\": \"%s\",\"%s\": \"%s\",\"%s\": [%s]}]}",
+              ReassignmentHandler.PLANS_KEY,
+              ReassignmentHandler.TOPIC_KEY,
+              topicName,
+              ReassignmentHandler.PARTITION_KEY,
+              "0",
+              ReassignmentHandler.TO_KEY,
+              nextBroker);
+
+      Assertions.assertEquals(Response.ACCEPT, handler.post(PostRequest.of(body)));
 
       Utils.sleep(Duration.ofSeconds(2));
       var reassignments = handler.get(Optional.of(topicName), Map.of());
@@ -82,19 +83,20 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
               .findFirst()
               .get();
 
-      Assertions.assertEquals(
-          Response.ACCEPT,
-          handler.post(
-              PostRequest.of(
-                  Map.of(
-                      ReassignmentHandler.TOPIC_KEY,
-                      topicName,
-                      ReassignmentHandler.PARTITION_KEY,
-                      "0",
-                      ReassignmentHandler.BROKER_KEY,
-                      String.valueOf(currentBroker),
-                      ReassignmentHandler.TO_KEY,
-                      nextPath))));
+      var body =
+          String.format(
+              "{\"%s\": [{\"%s\": \"%s\", \"%s\": \"%s\" ,\"%s\": \"%s\",\"%s\": \"%s\"}]}",
+              ReassignmentHandler.PLANS_KEY,
+              ReassignmentHandler.TOPIC_KEY,
+              topicName,
+              ReassignmentHandler.PARTITION_KEY,
+              "0",
+              ReassignmentHandler.BROKER_KEY,
+              currentBroker,
+              ReassignmentHandler.TO_KEY,
+              nextPath);
+
+      Assertions.assertEquals(Response.ACCEPT, handler.post(PostRequest.of(body)));
 
       Utils.sleep(Duration.ofSeconds(2));
       var reassignments = handler.get(Optional.of(topicName), Map.of());
@@ -106,4 +108,17 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
           admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).path());
     }
   }
+
+  @Test
+  void testBadRequest() {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ReassignmentHandler(admin);
+      admin.creator().topic(topicName).numberOfPartitions(1).create();
+      Utils.sleep(Duration.ofSeconds(3));
+      var body = "{\"plans\": []}";
+
+      Assertions.assertEquals(Response.BAD_REQUEST, handler.post(PostRequest.of(body)));
+    }
+  }
 }