Repository: skiptests/astraea
Pull request #348 (2022-05-16 16:16:02)

Enable to create topic by restful
新增一個方法讓使用者可以透過restful來建立topic，並且能指定partitions, replicas, and configuration


Patch:
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
index 68ada882..5fa68d89 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -11,17 +11,29 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 public final class Utils {
 
+  private static Throwable unpack(Throwable exception) {
+    Throwable current = exception;
+    while (current instanceof ExecutionException) {
+      current = current.getCause();
+    }
+    return current;
+  }
+
   public static <R> R handleException(Getter<R> getter) {
     try {
       return getter.get();
-    } catch (Exception e) {
-      throw new RuntimeException(e);
+    } catch (Throwable exception) {
+      var current = unpack(exception);
+      if (current instanceof RuntimeException) throw (RuntimeException) current;
+      if (current == null) throw new RuntimeException("unknown error");
+      throw new RuntimeException(current);
     }
   }
 
diff --git a/app/src/main/java/org/astraea/web/PostRequest.java b/app/src/main/java/org/astraea/web/PostRequest.java
index 962f0cc9..e3254cd2 100644
--- a/app/src/main/java/org/astraea/web/PostRequest.java
+++ b/app/src/main/java/org/astraea/web/PostRequest.java
@@ -6,40 +6,71 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 public interface PostRequest {
 
-  @SuppressWarnings("unchecked")
   static PostRequest of(HttpExchange exchange) throws IOException {
-    var jsonString = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
-    var raw =
-        ((Map<String, Object>) new Gson().fromJson(jsonString, Map.class))
+    return of(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8));
+  }
+
+  @SuppressWarnings("unchecked")
+  static PostRequest of(String json) {
+    return of(
+        ((Map<String, Object>) new Gson().fromJson(json, Map.class))
             .entrySet().stream()
                 .filter(e -> e.getValue() != null)
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toString()));
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> handleDouble(e.getValue()))));
+  }
+
+  static String handleDouble(Object obj) {
+    // 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));
+    }
+    return obj.toString();
+  }
+
+  static PostRequest of(Map<String, String> raw) {
     return () -> raw;
   }
 
   /** @return body represented by key-value */
   Map<String, String> raw();
 
+  default Optional<String> get(String key) {
+    return Optional.ofNullable(raw().get(key));
+  }
+
   default String value(String key) {
     var value = raw().get(key);
     if (value == null) throw new NoSuchElementException("the value for " + key + " is nonexistent");
     return value;
   }
 
+  default double doubleValue(String key, double defaultValue) {
+    return get(key).map(Double::parseDouble).orElse(defaultValue);
+  }
+
   default double doubleValue(String key) {
-    try {
-      return Double.parseDouble(value(key));
-    } catch (NumberFormatException e) {
-      throw new IllegalArgumentException(e);
-    }
+    return Double.parseDouble(value(key));
+  }
+
+  default int intValue(String key, int defaultValue) {
+    return get(key).map(Integer::parseInt).orElse(defaultValue);
   }
 
   default int intValue(String key) {
-    // the number in GSON is always DOUBLE>
-    return (int) doubleValue(key);
+    return Integer.parseInt(value(key));
+  }
+
+  default short shortValue(String key, short defaultValue) {
+    return get(key).map(Short::parseShort).orElse(defaultValue);
+  }
+
+  default short shortValue(String key) {
+    return Short.parseShort(value(key));
   }
 }
diff --git a/app/src/main/java/org/astraea/web/TopicHandler.java b/app/src/main/java/org/astraea/web/TopicHandler.java
index 210123e0..44e59c00 100644
--- a/app/src/main/java/org/astraea/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/web/TopicHandler.java
@@ -1,17 +1,23 @@
 package org.astraea.web;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.astraea.admin.Admin;
 import org.astraea.admin.Config;
 
 class TopicHandler implements Handler {
 
+  static final String TOPIC_NAME_KEY = "name";
+  static final String NUMBER_OF_PARTITIONS_KEY = "partitions";
+  static final String NUMBER_OF_REPLICAS_KEY = "replicas";
+
   private final Admin admin;
 
   TopicHandler(Admin admin) {
@@ -24,7 +30,11 @@ class TopicHandler implements Handler {
 
   @Override
   public JsonObject get(Optional<String> target, Map<String, String> queries) {
-    var topics = admin.topics(topicNames(target));
+    return get(topicNames(target));
+  }
+
+  private JsonObject get(Set<String> topicNames) {
+    var topics = admin.topics(topicNames);
     var replicas = admin.replicas(topics.keySet());
     var partitions =
         admin.offsets(topics.keySet()).entrySet().stream()
@@ -47,10 +57,39 @@ class TopicHandler implements Handler {
             .map(p -> new TopicInfo(p.getKey(), partitions.get(p.getKey()), p.getValue()))
             .collect(Collectors.toUnmodifiableList());
 
-    if (target.isPresent() && topicInfos.size() == 1) return topicInfos.get(0);
+    if (topicNames.size() == 1 && topicInfos.size() == 1) return topicInfos.get(0);
     return new Topics(topicInfos);
   }
 
+  static Map<String, String> remainingConfigs(PostRequest request) {
+    var configs = new HashMap<>(request.raw());
+    configs.remove(TOPIC_NAME_KEY);
+    configs.remove(NUMBER_OF_PARTITIONS_KEY);
+    configs.remove(NUMBER_OF_REPLICAS_KEY);
+    return configs;
+  }
+
+  @Override
+  public JsonObject post(PostRequest request) {
+    admin
+        .creator()
+        .topic(request.value(TOPIC_NAME_KEY))
+        .numberOfPartitions(request.intValue(NUMBER_OF_PARTITIONS_KEY, 1))
+        .numberOfReplicas(request.shortValue(NUMBER_OF_REPLICAS_KEY, (short) 1))
+        .configs(remainingConfigs(request))
+        .create();
+    if (admin.topicNames().contains(request.value(TOPIC_NAME_KEY))) {
+      try {
+        // if the topic creation is synced, we return the details.
+        return get(Set.of(request.value(TOPIC_NAME_KEY)));
+      } catch (UnknownTopicOrPartitionException e) {
+        // swallow
+      }
+    }
+    // Otherwise, return only name
+    return new TopicInfo(request.value(TOPIC_NAME_KEY), List.of(), Map.of());
+  }
+
   static class Topics implements JsonObject {
     final Collection<TopicInfo> topics;
 
@@ -65,11 +104,17 @@ class TopicHandler implements Handler {
     final Map<String, String> configs;
 
     private TopicInfo(String name, List<Partition> partitions, Config configs) {
+      this(
+          name,
+          partitions,
+          StreamSupport.stream(configs.spliterator(), false)
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+    }
+
+    private TopicInfo(String name, List<Partition> partitions, Map<String, String> configs) {
       this.name = name;
       this.partitions = partitions;
-      this.configs =
-          StreamSupport.stream(configs.spliterator(), false)
-              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+      this.configs = configs;
     }
   }
 
@@ -88,13 +133,13 @@ class TopicHandler implements Handler {
   }
 
   static class Replica implements JsonObject {
-    private final int broker;
-    private final long lag;
-    private final long size;
-    private final boolean leader;
-    private final boolean inSync;
-    private final boolean isFuture;
-    private final String path;
+    final int broker;
+    final long lag;
+    final long size;
+    final boolean leader;
+    final boolean inSync;
+    final boolean isFuture;
+    final String path;
 
     Replica(org.astraea.admin.Replica replica) {
       this(


Test patch:
diff --git a/app/src/test/java/org/astraea/UtilsTest.java b/app/src/test/java/org/astraea/UtilsTest.java
new file mode 100644
index 00000000..57ef6415
--- /dev/null
+++ b/app/src/test/java/org/astraea/UtilsTest.java
@@ -0,0 +1,36 @@
+package org.astraea;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class UtilsTest {
+
+  @Test
+  void testHandleException() {
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            Utils.handleException(
+                () -> {
+                  throw new ExecutionException(new IllegalArgumentException());
+                }));
+
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            Utils.handleException(
+                () -> {
+                  throw new IllegalArgumentException();
+                }));
+
+    Assertions.assertThrows(
+        RuntimeException.class,
+        () ->
+            Utils.handleException(
+                () -> {
+                  throw new IOException();
+                }));
+  }
+}
diff --git a/app/src/test/java/org/astraea/web/PostRequestTest.java b/app/src/test/java/org/astraea/web/PostRequestTest.java
index cf3193be..6fe5dfaa 100644
--- a/app/src/test/java/org/astraea/web/PostRequestTest.java
+++ b/app/src/test/java/org/astraea/web/PostRequestTest.java
@@ -11,7 +11,7 @@ import org.mockito.Mockito;
 public class PostRequestTest {
 
   @Test
-  void testParse() throws IOException {
+  void testParseHttpExchange() throws IOException {
     var input =
         new ByteArrayInputStream("{\"a\":\"b\",\"c\":123}".getBytes(StandardCharsets.UTF_8));
     var exchange = Mockito.mock(HttpExchange.class);
@@ -22,4 +22,20 @@ public class PostRequestTest {
     Assertions.assertEquals("b", request.raw().get("a"));
     Assertions.assertEquals(123, request.intValue("c"));
   }
+
+  @Test
+  void testHandleDouble() {
+    Assertions.assertEquals("10", PostRequest.handleDouble(10.00));
+    Assertions.assertEquals("10.01", PostRequest.handleDouble(10.01));
+    Assertions.assertEquals("xxxx", PostRequest.handleDouble("xxxx"));
+  }
+
+  @Test
+  void testParseJson() {
+    var request = PostRequest.of("{\"a\":1234, \"b\":3.34}");
+    Assertions.assertEquals(1234, request.intValue("a"));
+    Assertions.assertEquals(1234, request.shortValue("a"));
+    Assertions.assertEquals(1234.0, request.shortValue("a"));
+    Assertions.assertEquals(3.34, request.doubleValue("b"));
+  }
 }
diff --git a/app/src/test/java/org/astraea/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
index 0c563351..f5af95fb 100644
--- a/app/src/test/java/org/astraea/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
@@ -23,9 +23,16 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       var response =
           Assertions.assertInstanceOf(
               TopicHandler.Topics.class, handler.get(Optional.empty(), Map.of()));
-      Assertions.assertEquals(1, response.topics.size());
-      Assertions.assertEquals(topicName, response.topics.iterator().next().name);
-      Assertions.assertNotEquals(0, response.topics.iterator().next().configs.size());
+      Assertions.assertEquals(
+          1, response.topics.stream().filter(t -> t.name.equals(topicName)).count());
+      Assertions.assertNotEquals(
+          0,
+          response.topics.stream()
+              .filter(t -> t.name.equals(topicName))
+              .findFirst()
+              .get()
+              .configs
+              .size());
     }
   }
 
@@ -67,4 +74,75 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       Assertions.assertTrue(handler.topicNames(Optional.empty()).contains(topicName));
     }
   }
+
+  @Test
+  void testCreateTopic() {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new TopicHandler(admin);
+      var topicInfo =
+          Assertions.assertInstanceOf(
+              TopicHandler.TopicInfo.class,
+              handler.post(PostRequest.of(Map.of(TopicHandler.TOPIC_NAME_KEY, topicName))));
+      Assertions.assertEquals(topicName, topicInfo.name);
+    }
+  }
+
+  @Test
+  void testCreateTopicWithReplicas() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new TopicHandler(admin);
+      var topicInfo =
+          Assertions.assertInstanceOf(
+              TopicHandler.TopicInfo.class,
+              handler.post(
+                  PostRequest.of(
+                      Map.of(
+                          TopicHandler.TOPIC_NAME_KEY,
+                          topicName,
+                          TopicHandler.NUMBER_OF_PARTITIONS_KEY,
+                          "2",
+                          TopicHandler.NUMBER_OF_REPLICAS_KEY,
+                          "2",
+                          "segment.ms",
+                          "3000"))));
+      Assertions.assertEquals(topicName, topicInfo.name);
+
+      // the topic creation is not synced, so we have to wait the creation.
+      if (topicInfo.partitions.isEmpty()) {
+        TimeUnit.SECONDS.sleep(2);
+        var result = admin.replicas(Set.of(topicName));
+        Assertions.assertEquals(2, result.size());
+        result.values().forEach(replicas -> Assertions.assertEquals(2, replicas.size()));
+      } else {
+        Assertions.assertEquals(2, topicInfo.partitions.size());
+        Assertions.assertEquals(2, topicInfo.partitions.iterator().next().replicas.size());
+      }
+      Assertions.assertEquals(
+          "3000", admin.topics(Set.of(topicName)).get(topicName).value("segment.ms").get());
+    }
+  }
+
+  @Test
+  void testRemainingConfigs() {
+    Assertions.assertEquals(
+        0,
+        TopicHandler.remainingConfigs(
+                PostRequest.of(
+                    Map.of(
+                        TopicHandler.TOPIC_NAME_KEY,
+                        "abc",
+                        TopicHandler.NUMBER_OF_PARTITIONS_KEY,
+                        "2",
+                        TopicHandler.NUMBER_OF_REPLICAS_KEY,
+                        "2")))
+            .size());
+
+    Assertions.assertEquals(
+        1,
+        TopicHandler.remainingConfigs(
+                PostRequest.of(Map.of(TopicHandler.TOPIC_NAME_KEY, "abc", "key", "value")))
+            .size());
+  }
 }