Repository: skiptests/astraea
Pull request #437 (2022-06-28 14:56:48)

Web APIs for reassignments
新增Web APIs用來操作 reassignment，包含：

1) 查看目前的 `reassignment`
2) 移動partitions的部署節點
3) 移動replica的資料路徑


Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/Builder.java b/app/src/main/java/org/astraea/app/admin/Builder.java
index 911c2972..cbf82201 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -818,16 +818,19 @@ public class Builder {
     public void moveTo(Map<Integer, String> brokerFolders) {
       Utils.packException(
           () ->
-              admin.alterReplicaLogDirs(
-                  brokerFolders.entrySet().stream()
-                      .collect(
-                          Collectors.toMap(
-                              x ->
-                                  new TopicPartitionReplica(
-                                      partitions.iterator().next().topic(),
-                                      partitions.iterator().next().partition(),
-                                      x.getKey()),
-                              Map.Entry::getValue))));
+              admin
+                  .alterReplicaLogDirs(
+                      brokerFolders.entrySet().stream()
+                          .collect(
+                              Collectors.toMap(
+                                  x ->
+                                      new TopicPartitionReplica(
+                                          partitions.iterator().next().topic(),
+                                          partitions.iterator().next().partition(),
+                                          x.getKey()),
+                                  Map.Entry::getValue)))
+                  .all()
+                  .get());
     }
 
     @Override
diff --git a/app/src/main/java/org/astraea/app/admin/ReplicaCollie.java b/app/src/main/java/org/astraea/app/admin/ReplicaCollie.java
index fa7e92c5..e0d40a8e 100644
--- a/app/src/main/java/org/astraea/app/admin/ReplicaCollie.java
+++ b/app/src/main/java/org/astraea/app/admin/ReplicaCollie.java
@@ -210,7 +210,9 @@ public class ReplicaCollie {
       Integer broker) {
     pathMigrate.forEach(
         (tp, assignments) -> {
-          if (assignments.getValue().size() > 0)
+          // TODO: this code is a bit ugly, but it is fine as we are going to remove this class.
+          if (assignments.getValue().size() > 0
+              && !assignments.getValue().iterator().next().equals(UNKNOWN))
             admin
                 .migrator()
                 .partition(tp.topic(), tp.partition())
diff --git a/app/src/main/java/org/astraea/app/web/PostRequest.java b/app/src/main/java/org/astraea/app/web/PostRequest.java
index 05f75299..426940b6 100644
--- a/app/src/main/java/org/astraea/app/web/PostRequest.java
+++ b/app/src/main/java/org/astraea/app/web/PostRequest.java
@@ -17,9 +17,13 @@
 package org.astraea.app.web;
 
 import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
 import com.sun.net.httpserver.HttpExchange;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
@@ -41,7 +45,7 @@ public interface PostRequest {
   }
 
   static String handleDouble(Object obj) {
-    // the number in GSON is always DOUBLE>
+    // the number in GSON is always DOUBLE
     if (obj instanceof Double) {
       var value = (double) obj;
       if (value - Math.floor(value) == 0) return String.valueOf((long) Math.floor(value));
@@ -56,6 +60,14 @@ public interface PostRequest {
   /** @return body represented by key-value */
   Map<String, String> raw();
 
+  /**
+   * @param keys to check
+   * @return true if all keys has an associated value.
+   */
+  default boolean has(String... keys) {
+    return Arrays.stream(keys).allMatch(raw()::containsKey);
+  }
+
   default Optional<String> get(String key) {
     return Optional.ofNullable(raw().get(key));
   }
@@ -66,6 +78,16 @@ public interface PostRequest {
     return value;
   }
 
+  /**
+   * parse the value as a string array
+   *
+   * @param key to search value
+   * @return string array
+   */
+  default List<String> values(String key) {
+    return new Gson().fromJson(value(key), new TypeToken<ArrayList<String>>() {}.getType());
+  }
+
   default double doubleValue(String key, double defaultValue) {
     return get(key).map(Double::parseDouble).orElse(defaultValue);
   }
@@ -82,6 +104,14 @@ public interface PostRequest {
     return Integer.parseInt(value(key));
   }
 
+  default List<Integer> ints(String key) {
+    return values(key).stream()
+        // the number in GSON is always DOUBLE
+        .map(Double::valueOf)
+        .map(Double::intValue)
+        .collect(Collectors.toUnmodifiableList());
+  }
+
   default short shortValue(String key, short defaultValue) {
     return get(key).map(Short::parseShort).orElse(defaultValue);
   }
diff --git a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
new file mode 100644
index 00000000..862517c1
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.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.web;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
+
+public class ReassignmentHandler implements Handler {
+  static final String TOPIC_KEY = "topic";
+  static final String PARTITION_KEY = "partition";
+  static final String FROM_KEY = "from";
+  static final String TO_KEY = "to";
+  static final String BROKER_KEY = "broker";
+  private final Admin admin;
+
+  ReassignmentHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  @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.ints(TO_KEY));
+      return Response.ACCEPT;
+    }
+    return Response.BAD_REQUEST;
+  }
+
+  @Override
+  public Reassignments get(Optional<String> target, Map<String, String> queries) {
+    return new Reassignments(
+        admin.reassignments(Handler.compare(admin.topicNames(), target)).entrySet().stream()
+            .map(e -> new Reassignment(e.getKey(), e.getValue().from(), e.getValue().to()))
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  static class Location implements Response {
+    final int broker;
+    final String path;
+
+    Location(org.astraea.app.admin.Reassignment.Location location) {
+      this.broker = location.broker();
+      this.path = location.path();
+    }
+  }
+
+  static class Reassignment implements Response {
+    final String topicName;
+    final int partition;
+    final Collection<Location> from;
+    final Collection<Location> to;
+
+    Reassignment(
+        TopicPartition topicPartition,
+        Collection<org.astraea.app.admin.Reassignment.Location> from,
+        Collection<org.astraea.app.admin.Reassignment.Location> to) {
+      this.topicName = topicPartition.topic();
+      this.partition = topicPartition.partition();
+      this.from = from.stream().map(Location::new).collect(Collectors.toUnmodifiableList());
+      this.to = to.stream().map(Location::new).collect(Collectors.toUnmodifiableList());
+    }
+  }
+
+  static class Reassignments implements Response {
+    final Collection<Reassignment> reassignments;
+
+    Reassignments(Collection<Reassignment> reassignments) {
+      this.reassignments = reassignments;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/web/Response.java b/app/src/main/java/org/astraea/app/web/Response.java
index d483c448..a7d54d74 100644
--- a/app/src/main/java/org/astraea/app/web/Response.java
+++ b/app/src/main/java/org/astraea/app/web/Response.java
@@ -23,6 +23,7 @@ interface Response {
 
   Response OK = new ResponseImpl(200);
   Response ACCEPT = new ResponseImpl(202);
+  Response BAD_REQUEST = new ResponseImpl(400);
   Response NOT_FOUND = new ResponseImpl(404);
 
   static Response of(Exception exception) {
diff --git a/app/src/main/java/org/astraea/app/web/WebService.java b/app/src/main/java/org/astraea/app/web/WebService.java
index 4deeff96..bd128005 100644
--- a/app/src/main/java/org/astraea/app/web/WebService.java
+++ b/app/src/main/java/org/astraea/app/web/WebService.java
@@ -46,6 +46,7 @@ public class WebService {
     if (arg.needJmx())
       server.createContext("/beans", new BeanHandler(Admin.of(arg.configs()), arg.jmxPorts()));
     server.createContext("/records", new RecordHandler(arg.bootstrapServers()));
+    server.createContext("/reassignments", new ReassignmentHandler(Admin.of(arg.configs())));
     server.start();
   }
 
diff --git a/docs/web_server/README.md b/docs/web_server/README.md
index 55b508f5..b4461abb 100644
--- a/docs/web_server/README.md
+++ b/docs/web_server/README.md
@@ -22,3 +22,4 @@ Astraea 建立了一套 Web Server 服務，使用者可以透過簡易好上手
 - [/transactions](./web_api_transactions_chinese.md)
 - [/pipelines](./web_api_pipelines_chinese.md)
 - [/beans](./web_api_beans_chinese.md)
+- [/reassignments](./web_api_reassignments_chinese.md)
diff --git a/docs/web_server/web_api_reassignments_chinese.md b/docs/web_server/web_api_reassignments_chinese.md
new file mode 100644
index 00000000..18bee020
--- /dev/null
+++ b/docs/web_server/web_api_reassignments_chinese.md
@@ -0,0 +1,99 @@
+/reassignments
+===
+
+- [查詢所有 reassignments](#查詢所有-reassignments)
+- [變更 replicas 的節點部署](#變更-replicas-的節點部署)
+
+## 查詢所有 reassignments
+```shell
+GET /reassignments
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/reassignments
+```
+
+JSON Response 範例
+- `topicName`: 有 partitions 正在重新配置的 topic 名稱
+- `partition`: 有 replicas 正在重新配置的 partition id
+- `from`: replica 原本的位址
+  - `broker`: broker id
+  - `path`: 存放的資料夾路徑
+- `to`: replica 將來的位址
+```json
+{
+  "reassignments": [
+    {
+      "topicName": "chia",
+      "partition": 0,
+      "from": [
+        {
+          "broker": 1002,
+          "path": "/tmp/log-folder-0"
+        }
+      ],
+      "to": [
+        {
+          "broker": 1001,
+          "path": "/tmp/log-folder-1"
+        }
+      ]
+    }
+  ]
+}
+```
+
+## 變更 replicas 的節點部署
+
+```shell
+POST /reassignments
+```
+參數
+
+| 名稱        | 說明                                  | 預設  |
+|-----------|-------------------------------------|-----|
+| topic     | (必填) topic 名稱                       | 無   |
+| partition | (必填) partition id                   | 無 |
+| to        | (必填) 新的部署位置，以 broker id 為值，並且要是陣列型別 | 無 |
+
+cURL 範例
+
+將 chia-0 這個 partition 的部署改變成只部署在 broker = 1003 這台身上
+```shell
+curl -X POST http://localhost:8001/reassignments \
+    -H "Content-Type: application/json" \
+    -d '{
+    "topic": "chia", 
+    "partition": 0,
+    "to": [1003]
+    }' 
+```
+
+## 變更 replica 的資料路徑
+
+```shell
+POST /reassignments
+```
+參數
+
+| 名稱        | 說明                                   | 預設  |
+|-----------|--------------------------------------|-----|
+| topic     | (必填) topic 名稱                        | 無   |
+| partition | (必填) partition id                    | 無 |
+| broker    | (必填) replica 目前所在的節點                 | 無 |
+| to        | (必填) 新的用來存放資料的路徑，該資料夾必須是上述節點可用的資料夾位置 | 無 |
+
+cURL 範例
+
+將 broker = 1003 上的 chia-0 這個 partition 的資料存放路徑移動到 /tmp/data
+```shell
+curl -X POST http://localhost:8001/reassignments \
+    -H "Content-Type: application/json" \
+    -d '{
+    "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/PostRequestTest.java b/app/src/test/java/org/astraea/app/web/PostRequestTest.java
index b8698fb3..1e589b3f 100644
--- a/app/src/test/java/org/astraea/app/web/PostRequestTest.java
+++ b/app/src/test/java/org/astraea/app/web/PostRequestTest.java
@@ -54,4 +54,31 @@ public class PostRequestTest {
     Assertions.assertEquals(1234.0, request.shortValue("a"));
     Assertions.assertEquals(3.34, request.doubleValue("b"));
   }
+
+  @Test
+  void testStringArray() throws IOException {
+    var input =
+        new ByteArrayInputStream(
+            "{\"a\":\"b\",\"c\":[\"1\",\"2\"]}".getBytes(StandardCharsets.UTF_8));
+    var exchange = Mockito.mock(HttpExchange.class);
+    Mockito.when(exchange.getRequestBody()).thenReturn(input);
+
+    var request = PostRequest.of(exchange);
+    Assertions.assertEquals(2, request.values("c").size());
+    Assertions.assertEquals("1", request.values("c").get(0));
+    Assertions.assertEquals("2", request.values("c").get(1));
+  }
+
+  @Test
+  void testIntegerArray() throws IOException {
+    var input =
+        new ByteArrayInputStream("{\"a\":\"b\",\"c\":[1,2]}".getBytes(StandardCharsets.UTF_8));
+    var exchange = Mockito.mock(HttpExchange.class);
+    Mockito.when(exchange.getRequestBody()).thenReturn(input);
+
+    var request = PostRequest.of(exchange);
+    Assertions.assertEquals(2, request.ints("c").size());
+    Assertions.assertEquals(1, request.ints("c").get(0));
+    Assertions.assertEquals(2, request.ints("c").get(1));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
new file mode 100644
index 00000000..7b01f5a6
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.web;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ReassignmentHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testMigrateToAnotherBroker() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ReassignmentHandler(admin);
+      admin.creator().topic(topicName).numberOfPartitions(1).create();
+      TimeUnit.SECONDS.sleep(3);
+
+      var currentBroker =
+          admin.replicas(Set.of(topicName)).get(new TopicPartition(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 + "\"]"))));
+
+      TimeUnit.SECONDS.sleep(2);
+      var reassignments = handler.get(Optional.of(topicName), Map.of());
+      // the reassignment should be completed
+      Assertions.assertEquals(0, reassignments.reassignments.size());
+
+      Assertions.assertEquals(
+          nextBroker,
+          admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0).broker());
+    }
+  }
+
+  @Test
+  void testMigrateToAnotherPath() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ReassignmentHandler(admin);
+      admin.creator().topic(topicName).numberOfPartitions(1).create();
+      TimeUnit.SECONDS.sleep(3);
+
+      var currentReplica =
+          admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0);
+      var currentBroker = currentReplica.broker();
+      var currentPath = currentReplica.path();
+      var nextPath =
+          logFolders().get(currentBroker).stream()
+              .filter(p -> !p.equals(currentPath))
+              .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))));
+
+      TimeUnit.SECONDS.sleep(2);
+      var reassignments = handler.get(Optional.of(topicName), Map.of());
+      // the reassignment should be completed
+      Assertions.assertEquals(0, reassignments.reassignments.size());
+
+      Assertions.assertEquals(
+          nextPath,
+          admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0).path());
+    }
+  }
+}