opensource4you · chia7712 · Sep 17, 2022 · Sep 15, 2022
diff --git a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
@@ -16,10 +16,14 @@
  */
 package org.astraea.app.web;
 
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+
 import java.util.Collection;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.astraea.common.admin.Admin;
+import org.astraea.common.admin.Replica;
 import org.astraea.common.admin.TopicPartition;
 
 public class ReassignmentHandler implements Handler {
@@ -66,22 +70,29 @@ public Response post(Channel channel) {
 
   @Override
   public Reassignments get(Channel channel) {
+    var topics = Handler.compare(admin.topicNames(), channel.target());
+    var replicas = admin.replicas(topics);
     return new Reassignments(
-        admin
-            .reassignments(Handler.compare(admin.topicNames(), channel.target()))
-            .entrySet()
-            .stream()
-            .map(e -> new Reassignment(e.getKey(), e.getValue().from(), e.getValue().to()))
+        admin.reassignments(topics).entrySet().stream()
+            .map(
+                e ->
+                    toReassignment(
+                        e.getKey(),
+                        e.getValue().from(),
+                        e.getValue().to(),
+                        replicas.get(e.getKey())))
             .collect(Collectors.toUnmodifiableList()));
   }
 
   static class Location implements Response {
     final int broker;
     final String path;
+    final long size;
 
-    Location(org.astraea.common.admin.Reassignment.Location location) {
-      this.broker = location.broker();
-      this.path = location.dataFolder();
+    Location(int broker, String path, long size) {
+      this.broker = broker;
+      this.path = path;
+      this.size = size;
     }
   }
 
@@ -90,15 +101,19 @@ static class Reassignment implements Response {
     final int partition;
     final Collection<Location> from;
     final Collection<Location> to;
+    final String progress;
 
     Reassignment(
-        TopicPartition topicPartition,
-        Collection<org.astraea.common.admin.Reassignment.Location> from,
-        Collection<org.astraea.common.admin.Reassignment.Location> to) {
+        TopicPartition topicPartition, Collection<Location> from, Collection<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());
+      this.from = from;
+      this.to = to;
+
+      double fromSizeSum = from.stream().map(f -> f.size).reduce(0L, Long::sum);
+      double toSizeSum = to.stream().map(t -> t.size).reduce(0L, Long::sum);
+      double progress = fromSizeSum == 0 ? 0 : toSizeSum / fromSizeSum;
+      this.progress = progressInPercentage(progress);
     }
   }
 
@@ -109,4 +124,38 @@ static class Reassignments implements Response {
       this.reassignments = reassignments;
     }
   }
+
+  // visible for testing
+  static Reassignment toReassignment(
+      TopicPartition topicPartition,
+      Collection<org.astraea.common.admin.Reassignment.Location> from,
+      Collection<org.astraea.common.admin.Reassignment.Location> to,
+      Collection<Replica> replicas) {
+    return new Reassignment(
+        topicPartition,
+        from.stream()
+            .map(l -> new Location(l.broker(), l.dataFolder(), findReplica(replicas, l).size()))
+            .collect(Collectors.toUnmodifiableList()),
+        to.stream()
+            .map(l -> new Location(l.broker(), l.dataFolder(), findReplica(replicas, l).size()))
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  // visible for testing
+  static String progressInPercentage(double progress) {
+    // min(max(progress, 0), 1) is to force valid progress value is 0 < progress < 1
+    // in case something goes wrong (maybe compacting cause data size shrinking suddenly)
+    return String.format("%.2f%%", min(max(progress, 0), 1) * 100);
+  }
+
+  private static Replica findReplica(
+      Collection<Replica> replicas, org.astraea.common.admin.Reassignment.Location location) {
+    return replicas.stream()
+        .filter(
+            r ->
+                r.nodeInfo().id() == location.broker()
+                    && r.dataFolder().equals(location.dataFolder()))
+        .findFirst()
+        .orElseThrow(() -> new RuntimeException("replica not found"));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
@@ -16,10 +16,17 @@
  */
 package org.astraea.app.web;
 
+import static org.astraea.app.web.ReassignmentHandler.progressInPercentage;
+import static org.astraea.app.web.ReassignmentHandler.toReassignment;
+
 import java.time.Duration;
+import java.util.List;
 import java.util.Set;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
+import org.astraea.common.admin.NodeInfo;
+import org.astraea.common.admin.Reassignment;
+import org.astraea.common.admin.Replica;
 import org.astraea.common.admin.TopicPartition;
 import org.astraea.it.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -136,4 +143,72 @@ void testBadRequest() {
           Response.BAD_REQUEST, handler.post(Channel.ofRequest(PostRequest.of(body))));
     }
   }
+
+  @Test
+  void testToReassignment() {
+    var reassignment =
+        toReassignment(
+            TopicPartition.of("test", 0),
+            List.of(
+                new Reassignment.Location(1001, "/tmp/dir1"),
+                new Reassignment.Location(1002, "/tmp/dir1")),
+            List.of(
+                new Reassignment.Location(1001, "/tmp/dir2"),
+                new Reassignment.Location(1003, "/tmp/dir3")),
+            List.of(
+                fakeReplica(TopicPartition.of("test", 0), 1001, 200, "/tmp/dir1"),
+                fakeReplica(TopicPartition.of("test", 0), 1002, 200, "/tmp/dir1"),
+                fakeReplica(TopicPartition.of("test", 0), 1001, 20, "/tmp/dir2"),
+                fakeReplica(TopicPartition.of("test", 0), 1003, 30, "/tmp/dir3")));
+
+    Assertions.assertEquals(reassignment.topicName, "test");
+    Assertions.assertEquals(reassignment.partition, 0);
+
+    Assertions.assertEquals(reassignment.from.size(), 2);
+    var fromIterator = reassignment.from.iterator();
+    var from = fromIterator.next();
+    Assertions.assertEquals(from.broker, 1001);
+    Assertions.assertEquals(from.size, 200);
+    Assertions.assertEquals(from.path, "/tmp/dir1");
+    from = fromIterator.next();
+    Assertions.assertEquals(from.broker, 1002);
+    Assertions.assertEquals(from.size, 200);
+    Assertions.assertEquals(from.path, "/tmp/dir1");
+
+    Assertions.assertEquals(reassignment.to.size(), 2);
+    var toIterator = reassignment.to.iterator();
+    var to = toIterator.next();
+    Assertions.assertEquals(to.broker, 1001);
+    Assertions.assertEquals(to.size, 20);
+    Assertions.assertEquals(to.path, "/tmp/dir2");
+    to = toIterator.next();
+    Assertions.assertEquals(to.broker, 1003);
+    Assertions.assertEquals(to.size, 30);
+    Assertions.assertEquals(to.path, "/tmp/dir3");
+
+    Assertions.assertEquals(reassignment.progress, "12.50%");
+  }
+
+  private static Replica fakeReplica(
+      TopicPartition topicPartition, int brokerId, long size, String dataFolder) {
+    return Replica.of(
+        topicPartition.topic(),
+        topicPartition.partition(),
+        NodeInfo.of(brokerId, "", 12345),
+        0,
+        size,
+        true,
+        true,
+        true,
+        true,
+        true,
+        dataFolder);
+  }
+
+  @Test
+  void testProgressInPercentage() {
+    Assertions.assertEquals(progressInPercentage(-1.1), "0.00%");
+    Assertions.assertEquals(progressInPercentage(11.11), "100.00%");
+    Assertions.assertEquals(progressInPercentage(0.12345), "12.35%");
+  }
 }
diff --git a/docs/web_server/web_api_reassignments_chinese.md b/docs/web_server/web_api_reassignments_chinese.md
@@ -20,7 +20,10 @@ JSON Response 範例
 - `from`: replica 原本的位址
   - `broker`: broker id
   - `path`: 存放的資料夾路徑
+  - `size`: replica 大小，單位為 byte
 - `to`: replica 將來的位址
+- `progress`: 當前 replicas 搬移進度，以百分比顯示
+
 ```json
 {
   "reassignments": [
@@ -30,15 +33,18 @@ JSON Response 範例
       "from": [
         {
           "broker": 1002,
-          "path": "/tmp/log-folder-0"
+          "path": "/tmp/log-folder-0",
+          "size": 200
         }
       ],
       "to": [
         {
           "broker": 1001,
-          "path": "/tmp/log-folder-1"
+          "path": "/tmp/log-folder-1",
+          "size": 100
         }
-      ]
+      ],
+      "progress": "50.00%"
     }
   ]
 }