Repository: skiptests/astraea (java)

Issue:
Make `WebService` and standalone class
Context: https://github.com/skiptests/astraea/pull/1197#discussion_r1037338472

* 讓 `WebService` 變成一個真正的 Class Object，以 object instance 的方式使用他，而非呼叫 static method
* 讓 `WebService` object 釋出自己 binding 的 port 編號
* 讓 `WebService` 可以透過指定 port = 0 來自己挑選 port

Pull request description:
[WEB] Make WebService and standalone class
fix #1201

Patch:
diff --git a/app/src/main/java/org/astraea/app/web/BeanHandler.java b/app/src/main/java/org/astraea/app/web/BeanHandler.java
index 1eddecd2..9c19f25b 100644
--- a/app/src/main/java/org/astraea/app/web/BeanHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BeanHandler.java
@@ -17,6 +17,7 @@
 package org.astraea.app.web;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -27,9 +28,9 @@ import org.astraea.common.metrics.MBeanClient;
 
 public class BeanHandler implements Handler {
   private final Admin admin;
-  private final Function<String, Integer> jmxPorts;
+  private final Function<Integer, Optional<Integer>> jmxPorts;
 
-  BeanHandler(Admin admin, Function<String, Integer> jmxPorts) {
+  BeanHandler(Admin admin, Function<Integer, Optional<Integer>> jmxPorts) {
     this.admin = admin;
     this.jmxPorts = jmxPorts;
   }
@@ -42,7 +43,12 @@ public class BeanHandler implements Handler {
         .thenApply(
             brokers ->
                 brokers.stream()
-                    .map(b -> MBeanClient.jndi(b.host(), jmxPorts.apply(b.host())))
+                    .flatMap(
+                        b ->
+                            jmxPorts
+                                .apply(b.id())
+                                .map(port -> MBeanClient.jndi(b.host(), port))
+                                .stream())
                     .collect(Collectors.toList()))
         .thenApply(
             clients ->
diff --git a/app/src/main/java/org/astraea/app/web/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
index afd91b25..9394f494 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -34,6 +34,7 @@ import java.util.stream.Collectors;
 import org.astraea.common.Cache;
 import org.astraea.common.EnumInfo;
 import org.astraea.common.FutureUtils;
+import org.astraea.common.Lazy;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
 import org.astraea.common.admin.Partition;
@@ -64,23 +65,26 @@ public class RecordHandler implements Handler {
   private static final int MAX_CACHE_SIZE = 100;
   private static final Duration CACHE_EXPIRE_DURATION = Duration.ofMinutes(10);
 
-  final String bootstrapServers;
   // visible for testing
-  final Producer<byte[], byte[]> producer;
+  final Lazy<Producer<byte[], byte[]>> producer;
   private final Cache<String, Producer<byte[], byte[]>> transactionalProducerCache;
 
   final Admin admin;
 
-  RecordHandler(Admin admin, String bootstrapServers) {
+  RecordHandler(Admin admin) {
     this.admin = admin;
-    this.bootstrapServers = requireNonNull(bootstrapServers);
-    this.producer = Producer.builder().bootstrapServers(bootstrapServers).build();
+    this.producer =
+        Lazy.of(
+            () ->
+                Producer.builder()
+                    .bootstrapServers(admin.bootstrapServers().toCompletableFuture().join())
+                    .build());
     this.transactionalProducerCache =
         Cache.<String, Producer<byte[], byte[]>>builder(
                 transactionId ->
                     Producer.builder()
                         .config(ProducerConfigs.TRANSACTIONAL_ID_CONFIG, transactionId)
-                        .bootstrapServers(bootstrapServers)
+                        .bootstrapServers(admin.bootstrapServers().toCompletableFuture().join())
                         .buildTransactional())
             .maxCapacity(MAX_CACHE_SIZE)
             .expireAfterAccess(CACHE_EXPIRE_DURATION)
@@ -133,7 +137,7 @@ public class RecordHandler implements Handler {
             .orElse(SerDe.STRING)
             .deserializer;
     consumerBuilder
-        .bootstrapServers(bootstrapServers)
+        .bootstrapServers(admin.bootstrapServers().toCompletableFuture().join())
         .keyDeserializer(keyDeserializer)
         .valueDeserializer(valueDeserializer);
 
@@ -176,7 +180,10 @@ public class RecordHandler implements Handler {
       throw new IllegalArgumentException("records should contain at least one record");
 
     var producer =
-        postRequest.transactionId().map(transactionalProducerCache::get).orElse(this.producer);
+        postRequest
+            .transactionId()
+            .map(transactionalProducerCache::get)
+            .orElse(this.producer.get());
 
     var result =
         CompletableFuture.supplyAsync(
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 edf7264d..6da19831 100644
--- a/app/src/main/java/org/astraea/app/web/WebService.java
+++ b/app/src/main/java/org/astraea/app/web/WebService.java
@@ -19,41 +19,73 @@ package org.astraea.app.web;
 import com.beust.jcommander.Parameter;
 import com.sun.net.httpserver.HttpHandler;
 import com.sun.net.httpserver.HttpServer;
-import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.time.Duration;
 import java.util.Map;
-import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
+import org.astraea.common.argument.DurationField;
 import org.astraea.common.argument.NonNegativeIntegerField;
 import org.astraea.common.argument.StringMapField;
 
-public class WebService {
+public class WebService implements AutoCloseable {
 
-  public static void main(String[] args) throws Exception {
-    execute(org.astraea.common.argument.Argument.parse(new Argument(), args));
-  }
+  private final HttpServer server;
+  private final Admin admin;
 
-  private static void execute(Argument arg) throws IOException {
-    var server = HttpServer.create(new InetSocketAddress(arg.port), 0);
-    server.createContext("/topics", to(new TopicHandler(Admin.of(arg.configs()))));
-    server.createContext("/groups", to(new GroupHandler(Admin.of(arg.configs()))));
-    server.createContext("/brokers", to(new BrokerHandler(Admin.of(arg.configs()))));
-    server.createContext("/producers", to(new ProducerHandler(Admin.of(arg.configs()))));
-    server.createContext("/quotas", to(new QuotaHandler(Admin.of(arg.configs()))));
-    server.createContext("/transactions", to(new TransactionHandler(Admin.of(arg.configs()))));
-    if (arg.needJmx())
-      server.createContext("/beans", to(new BeanHandler(Admin.of(arg.configs()), arg.jmxPorts())));
-    server.createContext(
-        "/records", to(new RecordHandler(Admin.of(arg.configs()), arg.bootstrapServers())));
-    server.createContext("/reassignments", to(new ReassignmentHandler(Admin.of(arg.configs()))));
-    server.createContext(
-        "/balancer", to(new BalancerHandler(Admin.of(arg.configs()), arg.brokerJmxPorts())));
-    server.createContext("/throttles", to(new ThrottleHandler(Admin.of(arg.configs()))));
+  public WebService(Admin admin, int port, Function<Integer, Optional<Integer>> brokerIdToJmxPort) {
+    this.admin = admin;
+    server = Utils.packException(() -> HttpServer.create(new InetSocketAddress(port), 0));
+    server.createContext("/topics", to(new TopicHandler(admin)));
+    server.createContext("/groups", to(new GroupHandler(admin)));
+    server.createContext("/brokers", to(new BrokerHandler(admin)));
+    server.createContext("/producers", to(new ProducerHandler(admin)));
+    server.createContext("/quotas", to(new QuotaHandler(admin)));
+    server.createContext("/transactions", to(new TransactionHandler(admin)));
+    server.createContext("/beans", to(new BeanHandler(admin, brokerIdToJmxPort)));
+    server.createContext("/records", to(new RecordHandler(admin)));
+    server.createContext("/reassignments", to(new ReassignmentHandler(admin)));
+    server.createContext("/balancer", to(new BalancerHandler(admin, brokerIdToJmxPort)));
+    server.createContext("/throttles", to(new ThrottleHandler(admin)));
     server.start();
   }
 
+  public int port() {
+    return server.getAddress().getPort();
+  }
+
+  @Override
+  public void close() {
+    Utils.swallowException(admin::close);
+    server.stop(3);
+  }
+
+  public static void main(String[] args) throws Exception {
+    var arg = org.astraea.common.argument.Argument.parse(new Argument(), args);
+    Function<Integer, Optional<Integer>> brokerIdToPort =
+        id -> {
+          var r = Optional.ofNullable(arg.jmxPorts.get(String.valueOf(id))).map(Integer::parseInt);
+          if (r.isPresent()) return r;
+          if (arg.jmxPort > 0) return Optional.of(arg.jmxPort);
+          return Optional.empty();
+        };
+    try (var service = new WebService(Admin.of(arg.configs()), arg.port, brokerIdToPort)) {
+      if (arg.ttl == null) {
+        System.out.println("enter ctrl + c to terminate web service");
+        TimeUnit.MILLISECONDS.sleep(Long.MAX_VALUE);
+      } else {
+        System.out.println(
+            "this web service will get terminated automatically after "
+                + arg.ttl.toSeconds()
+                + " seconds");
+        TimeUnit.MILLISECONDS.sleep(arg.ttl.toMillis());
+      }
+    }
+  }
+
   private static HttpHandler to(Handler handler) {
     return exchange -> handler.handle(Channel.of(exchange));
   }
@@ -75,36 +107,17 @@ public class WebService {
 
     @Parameter(
         names = {"--jmx.ports"},
-        description = "Map: the JMX port for each node. For example: 192.168.50.2=19999",
-        validateWith = StringMapField.class,
-        converter = StringMapField.class)
-    Map<String, String> jmxPorts = Map.of();
-
-    @Parameter(
-        names = {"--broker.jmx.ports"},
         description =
             "Map: the JMX port for each broker. For example: 1024=19999 means for the broker with id 1024, its JMX port located at 19999 port",
         validateWith = StringMapField.class,
         converter = StringMapField.class)
-    Map<String, String> brokerJmxPorts = Map.of();
-
-    boolean needJmx() {
-      return jmxPort > 0 || !jmxPorts.isEmpty();
-    }
-
-    Function<String, Integer> jmxPorts() {
-      return name ->
-          Optional.of(jmxPorts.getOrDefault(name, String.valueOf(jmxPort)))
-              .map(Integer::valueOf)
-              .filter(i -> i > 0)
-              .orElseThrow(() -> new NoSuchElementException(name + " has no jmx port"));
-    }
+    Map<String, String> jmxPorts = Map.of();
 
-    Function<Integer, Optional<Integer>> brokerJmxPorts() {
-      return broker ->
-          Optional.of(brokerJmxPorts.getOrDefault(String.valueOf(broker), String.valueOf(jmxPort)))
-              .map(Integer::valueOf)
-              .filter(i -> i > 0);
-    }
+    @Parameter(
+        names = {"--ttl"},
+        description = "Duration: the life of web service",
+        validateWith = DurationField.class,
+        converter = DurationField.class)
+    Duration ttl = null;
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java b/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
index b0fd5124..4dfc9e98 100644
--- a/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
@@ -18,7 +18,7 @@ package org.astraea.app.web;
 
 import java.time.Duration;
 import java.util.Map;
-import java.util.NoSuchElementException;
+import java.util.Optional;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
 import org.astraea.it.RequireBrokerCluster;
@@ -27,29 +27,13 @@ import org.junit.jupiter.api.Test;
 
 public class BeanHandlerTest extends RequireBrokerCluster {
 
-  @Test
-  void testArgument() {
-    var arg = new WebService.Argument();
-    Assertions.assertFalse(arg.needJmx());
-    arg.jmxPort = 1000;
-    Assertions.assertTrue(arg.needJmx());
-    arg.jmxPort = -1;
-    arg.jmxPorts = Map.of("a", "100");
-    Assertions.assertTrue(arg.needJmx());
-    Assertions.assertEquals(100, arg.jmxPorts().apply("a"));
-    Assertions.assertThrows(NoSuchElementException.class, () -> arg.jmxPorts().apply("b"));
-
-    arg.jmxPort = 999;
-    Assertions.assertEquals(999, arg.jmxPorts().apply("b"));
-  }
-
   @Test
   void testBeans() {
     var topic = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(10).run().toCompletableFuture().join();
       Utils.sleep(Duration.ofSeconds(2));
-      var handler = new BeanHandler(admin, name -> jmxServiceURL().getPort());
+      var handler = new BeanHandler(admin, name -> Optional.of(jmxServiceURL().getPort()));
       var response =
           Assertions.assertInstanceOf(
               BeanHandler.NodeBeans.class, handler.get(Channel.EMPTY).toCompletableFuture().join());
diff --git a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
index 8aaed085..4f83e9db 100644
--- a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -69,7 +69,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
 
   @Test
   void testInvalidPost() {
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () ->
@@ -96,13 +96,13 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () ->
-            getRecordHandler()
+            new RecordHandler(Admin.of(bootstrapServers()))
                 .post(
                     Channel.ofRequest(
                         JsonConverter.defaultConverter().toJson(Map.of(TIMEOUT, "foo")))));
     Assertions.assertInstanceOf(
         RecordHandler.PostResponse.class,
-        getRecordHandler()
+        new RecordHandler(Admin.of(bootstrapServers()))
             .post(
                 Channel.ofRequest(
                     JsonConverter.defaultConverter()
@@ -124,7 +124,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.PostResponse.class,
-            getRecordHandler()
+            new RecordHandler(Admin.of(bootstrapServers()))
                 .post(
                     Channel.ofRequest(
                         "{\"records\":[{\"topic\":\"testPostRawString\", \"partition\":0,\"keySerializer\":\"string\",\"valueSerializer\":\"string\",\"key\":\"abc\",\"value\":\"abcd\"}]}"))
@@ -169,7 +169,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.PostResponse.class,
-            getRecordHandler()
+            new RecordHandler(Admin.of(bootstrapServers()))
                 .post(Channel.ofRequest(JsonConverter.defaultConverter().toJson(requestParams)))
                 .toCompletableFuture()
                 .join());
@@ -228,7 +228,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   @Test
   void testPostWithAsync() {
     var topic = Utils.randomString(10);
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var currentTimestamp = System.currentTimeMillis();
     var result =
         Assertions.assertInstanceOf(
@@ -255,7 +255,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 .join());
     Assertions.assertEquals(Response.ACCEPT, result);
 
-    handler.producer.flush();
+    handler.producer.get().flush();
 
     try (var consumer =
         Consumer.forTopics(Set.of(topic))
@@ -279,7 +279,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   @MethodSource("forTestSerializer")
   void testSerializer(String serializer, String actual, byte[] expected) {
     var topic = Utils.randomString(10);
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     Assertions.assertInstanceOf(
         RecordHandler.PostResponse.class,
         handler
@@ -336,7 +336,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
 
   @Test
   void testInvalidGet() {
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () -> handler.get(Channel.EMPTY).toCompletableFuture().join().code());
@@ -369,7 +369,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var topic = Utils.randomString(10);
     produceData(topic, 10);
 
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -393,7 +393,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var topic = Utils.randomString(10);
     produceData(topic, 10);
 
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -417,7 +417,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var topic = Utils.randomString(10);
     produceData(topic, 5);
 
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -462,7 +462,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
       producer.flush();
     }
 
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -483,7 +483,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var topic = Utils.randomString(10);
     produceData(topic, 10);
 
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -521,7 +521,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
       producer.flush();
     }
 
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -583,7 +583,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
               .build());
       producer.flush();
     }
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -636,7 +636,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
               .build());
       producer.flush();
     }
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var response =
         Assertions.assertInstanceOf(
             RecordHandler.GetResponse.class,
@@ -681,7 +681,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   @Test
   void testPostAndGet() {
     var topic = Utils.randomString(10);
-    var handler = getRecordHandler();
+    var handler = new RecordHandler(Admin.of(bootstrapServers()));
     var currentTimestamp = System.currentTimeMillis();
     Assertions.assertInstanceOf(
         RecordHandler.PostResponse.class,
@@ -742,9 +742,11 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   void testTimeout() {
     Assertions.assertThrows(
         IllegalArgumentException.class,
-        () -> getRecordHandler().get(Channel.ofQueries("test", Map.of(TIMEOUT, "foo"))));
+        () ->
+            new RecordHandler(Admin.of(bootstrapServers()))
+                .get(Channel.ofQueries("test", Map.of(TIMEOUT, "foo"))));
     var response =
-        getRecordHandler()
+        new RecordHandler(Admin.of(bootstrapServers()))
             .get(Channel.ofQueries("test", Map.of(TIMEOUT, "10s")))
             .toCompletableFuture()
             .join();
@@ -757,7 +759,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   void testDeleteParameter() {
     try (var admin = Admin.of(bootstrapServers())) {
       var topicName = Utils.randomString(10);
-      var handler = getRecordHandler();
+      var handler = new RecordHandler(Admin.of(bootstrapServers()));
       admin
           .creator()
           .topic(topicName)
@@ -795,7 +797,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       var topicName = Utils.randomString(10);
-      var handler = getRecordHandler();
+      var handler = new RecordHandler(Admin.of(bootstrapServers()));
       admin
           .creator()
           .topic(topicName)
@@ -879,7 +881,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       var topicName = Utils.randomString(10);
-      var handler = getRecordHandler();
+      var handler = new RecordHandler(Admin.of(bootstrapServers()));
       admin
           .creator()
           .topic(topicName)
@@ -936,7 +938,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       var topicName = Utils.randomString(10);
-      var handler = getRecordHandler();
+      var handler = new RecordHandler(Admin.of(bootstrapServers()));
       admin
           .creator()
           .topic(topicName)
@@ -992,7 +994,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   void testGetRecordsCommitOffsetWithGroupId() {
     var topic = Utils.randomString(10);
     var groupId = Utils.randomString(10);
-    var recordHandler = getRecordHandler();
+    var recordHandler = new RecordHandler(Admin.of(bootstrapServers()));
 
     Function<Boolean, RecordHandler.GetResponse> getRecords =
         needError ->
@@ -1042,7 +1044,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   @MethodSource("forTestGetRecordsCloseConsumer")
   void testGetRecordsCloseConsumer(Map<String, String> args) {
     var topic = Utils.randomString(10);
-    var recordHandler = getRecordHandler();
+    var recordHandler = new RecordHandler(Admin.of(bootstrapServers()));
 
     var response =
         Assertions.assertInstanceOf(
@@ -1075,17 +1077,13 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     Assertions.assertEquals(error2.getMessage(), "This consumer has already been closed.");
   }
 
-  private RecordHandler getRecordHandler() {
-    return new RecordHandler(Admin.of(bootstrapServers()), bootstrapServers());
-  }
-
   @Test
   void testRecords() {
     var exception = new IllegalArgumentException("hello");
     @SuppressWarnings({"unchecked", "resource"})
     Consumer<byte[], byte[]> consumer = Mockito.mock(Consumer.class);
     Mockito.when(consumer.poll(Mockito.anyInt(), Mockito.any())).thenThrow(exception);
-    var recordHandler = getRecordHandler();
+    var recordHandler = new RecordHandler(Admin.of(bootstrapServers()));
     Assertions.assertEquals(
         exception,
         Assertions.assertThrows(
diff --git a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
index a21a2715..a51ecf0c 100644
--- a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
@@ -22,11 +22,10 @@ import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletionException;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.common.Utils;
@@ -58,40 +57,26 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       topic.replicas = replicas;
       request.topics = List.of(topic);
 
-      var serviceThread = Executors.newSingleThreadExecutor();
-      var servicePort = ThreadLocalRandom.current().nextInt(10000, 65536);
-      serviceThread.submit(
-          () ->
-              Utils.packException(
-                  () ->
-                      WebService.main(
-                          new String[] {
-                            "--port",
-                            String.valueOf(servicePort),
-                            "--bootstrap.servers",
-                            bootstrapServers()
-                          })));
-      Response<TopicHandler.Topics> response =
-          HttpExecutor.builder()
-              .build()
-              .post(
-                  "http://localhost:" + servicePort + "/topics",
-                  request,
-                  TypeRef.of(TopicHandler.Topics.class))
-              .toCompletableFuture()
-              .join();
-      Utils.sleep(Duration.ofSeconds(1));
+      try (var service = new WebService(Admin.of(bootstrapServers()), 0, id -> Optional.empty())) {
+        Response<TopicHandler.Topics> response =
+            HttpExecutor.builder()
+                .build()
+                .post(
+                    "http://localhost:" + service.port() + "/topics",
+                    request,
+                    TypeRef.of(TopicHandler.Topics.class))
+                .toCompletableFuture()
+                .join();
+        Utils.sleep(Duration.ofSeconds(1));
 
-      var clusterInfo = admin.clusterInfo(Set.of(topicName)).toCompletableFuture().join();
-      Assertions.assertEquals(200, response.statusCode());
-      Assertions.assertEquals(1, response.body().topics.size());
-      Assertions.assertEquals(partitions, response.body().topics.get(0).partitions.size());
-      Assertions.assertEquals(Set.of(topicName), clusterInfo.topics());
-      Assertions.assertEquals(partitions, clusterInfo.topicPartitions().size());
-      Assertions.assertEquals(partitions * replicas, clusterInfo.replicas().size());
-      serviceThread.shutdown();
-      Assertions.assertTrue(
-          Utils.packException(() -> serviceThread.awaitTermination(10, TimeUnit.SECONDS)));
+        var clusterInfo = admin.clusterInfo(Set.of(topicName)).toCompletableFuture().join();
+        Assertions.assertEquals(200, response.statusCode());
+        Assertions.assertEquals(1, response.body().topics.size());
+        Assertions.assertEquals(partitions, response.body().topics.get(0).partitions.size());
+        Assertions.assertEquals(Set.of(topicName), clusterInfo.topics());
+        Assertions.assertEquals(partitions, clusterInfo.topicPartitions().size());
+        Assertions.assertEquals(partitions * replicas, clusterInfo.replicas().size());
+      }
     }
   }
 
diff --git a/app/src/test/java/org/astraea/app/web/WebServiceTest.java b/app/src/test/java/org/astraea/app/web/WebServiceTest.java
index 0dca754a..84527c2c 100644
--- a/app/src/test/java/org/astraea/app/web/WebServiceTest.java
+++ b/app/src/test/java/org/astraea/app/web/WebServiceTest.java
@@ -16,9 +16,13 @@
  */
 package org.astraea.app.web;
 
+import java.util.Optional;
+import org.astraea.common.admin.Admin;
 import org.astraea.common.argument.Argument;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.mockito.Mockito;
 
 public class WebServiceTest {
 
@@ -31,4 +35,11 @@ public class WebServiceTest {
     Assertions.assertEquals("localhost", argument.bootstrapServers());
     Assertions.assertEquals(65535, argument.port);
   }
+
+  @Timeout(5)
+  @Test
+  void testClose() {
+    var web = new WebService(Mockito.mock(Admin.class), 0, id -> Optional.empty());
+    web.close();
+  }
 }