Repository: skiptests/astraea
Pull request #330 (2022-05-12 11:42:11)

Add web service to inspect kafka cluster
這是一個獨立的新應用程式，目的是將一些非常需要觀察的叢集資訊透過restful APIs曝露出來，目前專案原有的工具都是將資料印到console上面，這在與下游整合上不太容易，因此我們要改成restful APIs的形式。這隻PR作為起始動作包含幾個目標：

1. 盡量不引入太多第三方函式庫，整個web service目前只額外用到google json
2. 目前支援查詢`topics` and `groups`
3. `start_kafka_tool.sh`腳本有配合修改讓web service所監聽的port可以開放

之後會逐步擴充web service能查詢的項目，最終目標要取代掉現有的`offsets`和`monitor`

---

範例一：查詢topic: testPerformance-1652350196001

```
GET http://localhost:8001/topics/testPerformance-1652350196001
```
```json
{
    "name": "testPerformance-1652350196001",
    "partitions": [
        {
            "partitionId": 6,
            "earliest": 0,
            "latest": 149844,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78885381,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-2"
                }
            ]
        },
        {
            "partitionId": 2,
            "earliest": 0,
            "latest": 149888,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78853647,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-1"
                }
            ]
        },
        {
            "partitionId": 1,
            "earliest": 0,
            "latest": 149498,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78845590,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-2"
                }
            ]
        },
        {
            "partitionId": 7,
            "earliest": 0,
            "latest": 149621,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78819273,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-2"
                }
            ]
        },
        {
            "partitionId": 3,
            "earliest": 0,
            "latest": 146126,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 76988968,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-1"
                }
            ]
        },
        {
            "partitionId": 8,
            "earliest": 0,
            "latest": 149658,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78827676,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-1"
                }
            ]
        },
        {
            "partitionId": 4,
            "earliest": 0,
            "latest": 149447,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78847277,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-0"
                }
            ]
        },
        {
            "partitionId": 9,
            "earliest": 0,
            "latest": 149744,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78832251,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-0"
                }
            ]
        },
        {
            "partitionId": 5,
            "earliest": 0,
            "latest": 149325,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78844430,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-0"
                }
            ]
        },
        {
            "partitionId": 0,
            "earliest": 0,
            "latest": 149761,
            "replicas": [
                {
                    "broker": 1001,
                    "lag": 0,
                    "size": 78895578,
                    "leader": true,
                    "inSync": true,
                    "isFuture": false,
                    "path": "/tmp/log-folder-1"
                }
            ]
        }
    ],
    "configs": {
        "compression.type": "producer",
        "leader.replication.throttled.replicas": "",
        "message.downconversion.enable": "true",
        "min.insync.replicas": "1",
        "segment.jitter.ms": "0",
        "cleanup.policy": "delete",
        "flush.ms": "9223372036854775807",
        "follower.replication.throttled.replicas": "",
        "segment.bytes": "1073741824",
        "retention.ms": "604800000",
        "flush.messages": "9223372036854775807",
        "message.format.version": "3.0-IV1",
        "max.compaction.lag.ms": "9223372036854775807",
        "file.delete.delay.ms": "60000",
        "max.message.bytes": "1048588",
        "min.compaction.lag.ms": "0",
        "message.timestamp.type": "CreateTime",
        "preallocate": "false",
        "min.cleanable.dirty.ratio": "0.5",
        "index.interval.bytes": "4096",
        "unclean.leader.election.enable": "false",
        "retention.bytes": "-1",
        "delete.retention.ms": "86400000",
        "segment.ms": "604800000",
        "message.timestamp.difference.max.ms": "9223372036854775807",
        "segment.index.bytes": "10485760"
    }
}
```

chia7712: @garyparrot 感謝審查，已經修正相關意見

Patch:
diff --git a/README.md b/README.md
index c11bbaaf..ad2a0403 100644
--- a/README.md
+++ b/README.md
@@ -439,3 +439,32 @@ $ ./gradlew run --args="monitor --bootstrap.servers 192.168.103.39:9092"
 3. --prop.file: the path to a file that containing the properties to be passed to kafka admin.
 4. --topic: topics to track (default: track all non-synced partition by default)
 5. --track: keep track even if all the replicas are synced. Also attempts to discover any non-synced replicas. (default: false)
+
+### Web service to inspect details of your Kafka data
+
+1. --bootstrap.servers: the server to connect to
+2. --port: the port used by web server
+
+```shell
+./docker/start_kafka_tool.sh web --bootstrap.servers 192.168.50.178:19993 --port 12345"
+```
+
+## Query all topics 
+```shell
+GET http://localhost:12345/topics
+```
+
+## Query single topic
+```shell
+GET http://localhost:12345/topics/t0
+```
+
+## Query all groups
+```shell
+GET http://localhost:12345/groups
+```
+
+## Query single group
+```shell
+GET http://localhost:12345/groups/g1
+```
\ No newline at end of file
diff --git a/app/build.gradle b/app/build.gradle
index 2847901d..ad38b64f 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -16,6 +16,7 @@ def versions = [
     jcommander: project.properties['jcommander.version'] ?: "1.81",
     slf4j: project.properties['slf4j.version'] ?: "1.7.32",
     zookeeper: project.properties['zookeeper.version'] ?: "3.6.3",
+    gson: project.properties['gson.version'] ?: "2.9.0",
 ]
 
 dependencies {
@@ -30,6 +31,7 @@ dependencies {
     implementation "com.beust:jcommander:${versions["jcommander"]}"
     // we don't use slf4j actually, and it is used by kafka so we swallow the log.
     implementation "org.slf4j:slf4j-nop:${versions["slf4j"]}"
+    implementation "com.google.code.gson:gson:${versions["gson"]}"
 }
 
 application {
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index 23312892..c936550a 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -12,6 +12,7 @@ import org.astraea.topic.ReplicaCollie;
 import org.astraea.topic.ReplicaSyncingMonitor;
 import org.astraea.topic.TopicExplorer;
 import org.astraea.topic.cost.PartitionScore;
+import org.astraea.web.WebService;
 
 public class App {
   private static final Map<String, Class<?>> MAIN_CLASSES =
@@ -22,7 +23,8 @@ public class App {
           "score", PartitionScore.class,
           "performance", Performance.class,
           "monitor", ReplicaSyncingMonitor.class,
-          "automation", Automation.class);
+          "automation", Automation.class,
+          "web", WebService.class);
 
   static void execute(Map<String, Class<?>> mains, List<String> args) throws Throwable {
 
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index f68a4738..5b17c376 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -44,6 +44,11 @@ public interface TopicAdmin extends Closeable {
    */
   Map<TopicPartition, Offset> offsets(Set<String> topics);
 
+  /** @return all consumer groups */
+  default Map<String, ConsumerGroup> consumerGroup() {
+    return consumerGroup(Set.of());
+  }
+
   /**
    * @param consumerGroupNames consumer group names. if empty set given, every consume group will
    *     return.
diff --git a/app/src/main/java/org/astraea/web/ErrorObject.java b/app/src/main/java/org/astraea/web/ErrorObject.java
new file mode 100644
index 00000000..0820f70f
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/ErrorObject.java
@@ -0,0 +1,13 @@
+package org.astraea.web;
+
+import java.util.NoSuchElementException;
+
+class ErrorObject implements JsonObject {
+  final int code;
+  final String message;
+
+  ErrorObject(Exception exception) {
+    this.code = exception instanceof NoSuchElementException ? 404 : 400;
+    this.message = exception.getMessage();
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/GroupHandler.java b/app/src/main/java/org/astraea/web/GroupHandler.java
new file mode 100644
index 00000000..aab2050b
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/GroupHandler.java
@@ -0,0 +1,127 @@
+package org.astraea.web;
+
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.topic.TopicAdmin;
+
+public class GroupHandler implements Handler {
+
+  private final TopicAdmin admin;
+
+  GroupHandler(TopicAdmin admin) {
+    this.admin = admin;
+  }
+
+  @Override
+  public JsonObject response(Optional<String> target, Map<String, String> queries) {
+    Predicate<Map.Entry<String, ?>> groupFilter =
+        e -> target.stream().allMatch(t -> t.equals(e.getKey()));
+    var topics = admin.topicNames();
+    var consumerGroups = admin.consumerGroup();
+    var offsets = admin.offsets(topics);
+
+    var groups =
+        consumerGroups.entrySet().stream()
+            .filter(groupFilter)
+            .map(
+                cgAndTp ->
+                    new Group(
+                        cgAndTp.getKey(),
+                        cgAndTp.getValue().assignment().entrySet().stream()
+                            .map(
+                                entry ->
+                                    new Member(
+                                        entry.getKey().memberId(),
+                                        entry.getKey().groupInstanceId(),
+                                        entry.getKey().clientId(),
+                                        entry.getKey().host(),
+                                        entry.getValue().stream()
+                                            .map(
+                                                tp ->
+                                                    offsets.containsKey(tp)
+                                                            && cgAndTp
+                                                                .getValue()
+                                                                .consumeProgress()
+                                                                .containsKey(tp)
+                                                        ? Optional.of(
+                                                            new OffsetProgress(
+                                                                tp.topic(),
+                                                                tp.partition(),
+                                                                offsets.get(tp).earliest(),
+                                                                cgAndTp
+                                                                    .getValue()
+                                                                    .consumeProgress()
+                                                                    .get(tp),
+                                                                offsets.get(tp).latest()))
+                                                        : Optional.<OffsetProgress>empty())
+                                            .filter(Optional::isPresent)
+                                            .map(Optional::get)
+                                            .collect(Collectors.toUnmodifiableList())))
+                            .collect(Collectors.toUnmodifiableList())))
+            .collect(Collectors.toUnmodifiableList());
+
+    if (target.isPresent() && groups.size() == 1) return groups.get(0);
+    else if (target.isPresent())
+      throw new NoSuchElementException("group: " + target.get() + " does not exist");
+    else return new Groups(groups);
+  }
+
+  static class OffsetProgress implements JsonObject {
+    final String topicName;
+    final int partitionId;
+    final long earliest;
+    final long current;
+    final long latest;
+
+    OffsetProgress(String topicName, int partitionId, long earliest, long current, long latest) {
+      this.topicName = topicName;
+      this.partitionId = partitionId;
+      this.earliest = earliest;
+      this.current = current;
+      this.latest = latest;
+    }
+  }
+
+  static class Member implements JsonObject {
+    private final String memberId;
+    private final Optional<String> groupInstanceId;
+    private final String clientId;
+    private final String host;
+    private final List<OffsetProgress> offsetProgress;
+
+    Member(
+        String memberId,
+        Optional<String> groupInstanceId,
+        String clientId,
+        String host,
+        List<OffsetProgress> offsetProgress) {
+      this.memberId = memberId;
+      this.groupInstanceId = groupInstanceId;
+      this.clientId = clientId;
+      this.host = host;
+      this.offsetProgress = offsetProgress;
+    }
+  }
+
+  static class Group implements JsonObject {
+    final String groupId;
+    final List<Member> members;
+
+    Group(String groupId, List<Member> members) {
+      this.groupId = groupId;
+      this.members = members;
+    }
+  }
+
+  static class Groups implements JsonObject {
+    final List<Group> groups;
+
+    Groups(List<Group> groups) {
+      this.groups = groups;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/Handler.java b/app/src/main/java/org/astraea/web/Handler.java
new file mode 100644
index 00000000..6d3cf54b
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/Handler.java
@@ -0,0 +1,59 @@
+package org.astraea.web;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+interface Handler extends HttpHandler {
+
+  default JsonObject response(HttpExchange exchange) {
+    try {
+      return response(
+          parseTarget(exchange.getRequestURI()), parseQueries(exchange.getRequestURI()));
+    } catch (Exception e) {
+      e.printStackTrace();
+      return new ErrorObject(e);
+    }
+  }
+
+  @Override
+  default void handle(HttpExchange exchange) throws IOException {
+    var response = response(exchange);
+    var responseData = response.json().getBytes(StandardCharsets.UTF_8);
+    exchange
+        .getResponseHeaders()
+        .set("Content-Type", String.format("application/json; charset=%s", StandardCharsets.UTF_8));
+    exchange.sendResponseHeaders(
+        response instanceof ErrorObject ? ((ErrorObject) response).code : 200, responseData.length);
+    try (var os = exchange.getResponseBody()) {
+      os.write(responseData);
+    }
+  }
+
+  static Optional<String> parseTarget(URI uri) {
+    var allPaths =
+        Arrays.stream(uri.getPath().split("/"))
+            .map(String::trim)
+            .filter(s -> !s.isEmpty() && !s.isBlank())
+            .collect(Collectors.toUnmodifiableList());
+    // form: /resource/target
+    if (allPaths.size() == 1) return Optional.empty();
+    else if (allPaths.size() == 2) return Optional.of(allPaths.get(1));
+    else throw new IllegalArgumentException("unsupported url: " + uri);
+  }
+
+  static Map<String, String> parseQueries(URI uri) {
+    if (uri.getQuery() == null || uri.getQuery().isEmpty()) return Map.of();
+    return Arrays.stream(uri.getQuery().split("&"))
+        .filter(pair -> pair.split("=").length == 2)
+        .collect(Collectors.toMap(p -> p.split("=")[0], p -> p.split("=")[1]));
+  }
+
+  JsonObject response(Optional<String> target, Map<String, String> queries);
+}
diff --git a/app/src/main/java/org/astraea/web/JsonObject.java b/app/src/main/java/org/astraea/web/JsonObject.java
new file mode 100644
index 00000000..f3d6fe0c
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/JsonObject.java
@@ -0,0 +1,9 @@
+package org.astraea.web;
+
+import com.google.gson.Gson;
+
+interface JsonObject {
+  default String json() {
+    return new Gson().toJson(this);
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/TopicHandler.java b/app/src/main/java/org/astraea/web/TopicHandler.java
new file mode 100644
index 00000000..bab43f8d
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/TopicHandler.java
@@ -0,0 +1,129 @@
+package org.astraea.web;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import org.astraea.topic.TopicAdmin;
+import org.astraea.topic.TopicConfig;
+
+class TopicHandler implements Handler {
+
+  private final TopicAdmin admin;
+
+  TopicHandler(TopicAdmin admin) {
+    this.admin = admin;
+  }
+
+  @Override
+  public JsonObject response(Optional<String> target, Map<String, String> queries) {
+    Predicate<Map.Entry<String, ?>> topicFilter =
+        e -> target.stream().allMatch(t -> t.equals(e.getKey()));
+    var topics = admin.topics();
+    var replicas = admin.replicas(topics.keySet());
+    var partitions =
+        admin.offsets(topics.keySet()).entrySet().stream()
+            .collect(
+                Collectors.groupingBy(
+                    e -> e.getKey().topic(),
+                    Collectors.mapping(
+                        e ->
+                            new Partition(
+                                e.getKey().partition(),
+                                e.getValue().earliest(),
+                                e.getValue().latest(),
+                                replicas.get(e.getKey()).stream()
+                                    .map(Replica::new)
+                                    .collect(Collectors.toUnmodifiableList())),
+                        Collectors.toList())));
+
+    var topicInfos =
+        topics.entrySet().stream()
+            .filter(topicFilter)
+            .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);
+    else if (target.isPresent())
+      throw new NoSuchElementException("topic: " + target.get() + " does not exist");
+    else return new Topics(topicInfos);
+  }
+
+  static class Topics implements JsonObject {
+    final Collection<TopicInfo> topics;
+
+    private Topics(Collection<TopicInfo> topics) {
+      this.topics = topics;
+    }
+  }
+
+  static class TopicInfo implements JsonObject {
+    final String name;
+    final List<Partition> partitions;
+    final Map<String, String> configs;
+
+    private TopicInfo(String name, List<Partition> partitions, TopicConfig configs) {
+      this.name = name;
+      this.partitions = partitions;
+      this.configs =
+          StreamSupport.stream(configs.spliterator(), false)
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+  }
+
+  static class Partition implements JsonObject {
+    final int id;
+    final long earliest;
+    final long latest;
+    final List<Replica> replicas;
+
+    Partition(int id, long earliest, long latest, List<Replica> replicas) {
+      this.id = id;
+      this.earliest = earliest;
+      this.latest = latest;
+      this.replicas = replicas;
+    }
+  }
+
+  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;
+
+    Replica(org.astraea.topic.Replica replica) {
+      this(
+          replica.broker(),
+          replica.lag(),
+          replica.size(),
+          replica.leader(),
+          replica.inSync(),
+          replica.isFuture(),
+          replica.path());
+    }
+
+    Replica(
+        int broker,
+        long lag,
+        long size,
+        boolean leader,
+        boolean inSync,
+        boolean isFuture,
+        String path) {
+      this.broker = broker;
+      this.lag = lag;
+      this.size = size;
+      this.leader = leader;
+      this.inSync = inSync;
+      this.isFuture = isFuture;
+      this.path = path;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
new file mode 100644
index 00000000..7ca1c001
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -0,0 +1,31 @@
+package org.astraea.web;
+
+import com.beust.jcommander.Parameter;
+import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import org.astraea.argument.NonNegativeIntegerField;
+import org.astraea.topic.TopicAdmin;
+
+public class WebService {
+
+  public static void main(String[] args) throws Exception {
+    execute(org.astraea.argument.Argument.parse(new Argument(), args));
+  }
+
+  private static void execute(Argument arg) throws IOException {
+    var server = HttpServer.create(new InetSocketAddress(arg.port), 0);
+    server.createContext("/topics", new TopicHandler(TopicAdmin.of(arg.bootstrapServers())));
+    server.createContext("/groups", new GroupHandler(TopicAdmin.of(arg.bootstrapServers())));
+    server.start();
+  }
+
+  static class Argument extends org.astraea.argument.Argument {
+    @Parameter(
+        names = {"--port"},
+        description = "Integer: the port to bind",
+        validateWith = NonNegativeIntegerField.class,
+        converter = NonNegativeIntegerField.class)
+    int port = 8001;
+  }
+}
diff --git a/docker/start_kafka_tool.sh b/docker/start_kafka_tool.sh
index afeb00e5..93284729 100755
--- a/docker/start_kafka_tool.sh
+++ b/docker/start_kafka_tool.sh
@@ -1,6 +1,5 @@
 #!/bin/bash
 
-
 declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
 source $DOCKER_FOLDER/docker_build_common.sh
 
@@ -10,6 +9,8 @@ declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/kafka-tool}
 declare -r IMAGE_NAME="$REPO:$VERSION"
 declare -r DOCKERFILE=$DOCKER_FOLDER/kafka_tool.dockerfile
 declare -r JMX_PORT=${JMX_PORT:-"$(getRandomPort)"}
+# for web service
+declare -r WEB_PORT=${JMX_PORT:-"$(getRandomPort)"}
 declare -r JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false \
                      -Dcom.sun.management.jmxremote.port=$JMX_PORT -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT -Djava.rmi.server.hostname=$ADDRESS"
 declare -r HEAP_OPTS="${HEAP_OPTS:-"-Xmx2G -Xms2G"}"
@@ -68,8 +69,34 @@ WORKDIR /opt/astraea
 function runContainer() {
   local args=$1
   echo "JMX address: $ADDRESS:$JMX_PORT"
+
+  # web service needs to bind a port to expose Restful APIs, so we have to open a port of container
+  need_to_bind_web=""
+  if [[ "$args" == web* ]]; then
+    sentence=($args)
+    defined_port="false"
+    # use random port by default
+    web_port="$WEB_PORT"
+    for word in "${sentence[@]}"; do
+      # user has pre-defined port, so we will replace the random port by this one in next loop
+      if [[ "$word" == "--port" ]]; then
+        defined_port="true"
+      fi
+      # this element must be port
+      if [[ "$defined_port" == "true" ]]; then
+        web_port="$word"
+      fi
+    done
+    # manually add "--port" to make sure web service bind on random port we pass
+    if [[ "$defined_port" == "false" ]]; then
+      args="$args --port $web_port"
+    fi
+    need_to_bind_web="-p $web_port:$web_port"
+  fi
+
   docker run --rm --init \
     -p $JMX_PORT:$JMX_PORT \
+    $need_to_bind_web \
     "$IMAGE_NAME" \
     /bin/bash -c "java $JMX_OPTS $HEAP_OPTS -jar /opt/astraea/app.jar $args"
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
new file mode 100644
index 00000000..85f3879f
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
@@ -0,0 +1,78 @@
+package org.astraea.web;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.astraea.Utils;
+import org.astraea.consumer.Consumer;
+import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class GroupHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testListGroups() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    var groupId = Utils.randomString(10);
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      admin.creator().topic(topicName).create();
+      TimeUnit.SECONDS.sleep(3);
+
+      try (var consumer =
+          Consumer.builder()
+              .groupId(groupId)
+              .topics(Set.of(topicName))
+              .bootstrapServers(bootstrapServers())
+              .build()) {
+        Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+        var handler = new GroupHandler(admin);
+        var response =
+            Assertions.assertInstanceOf(
+                GroupHandler.Groups.class, handler.response(Optional.empty(), Map.of()));
+        Assertions.assertEquals(1, response.groups.size());
+        Assertions.assertEquals(groupId, response.groups.iterator().next().groupId);
+        Assertions.assertEquals(1, response.groups.iterator().next().members.size());
+      }
+    }
+  }
+
+  @Test
+  void testQueryNonexistentGroup() {
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      var handler = new GroupHandler(admin);
+      var exception =
+          Assertions.assertThrows(
+              NoSuchElementException.class,
+              () -> handler.response(Optional.of("unknown"), Map.of()));
+      Assertions.assertTrue(exception.getMessage().contains("unknown"));
+    }
+  }
+
+  @Test
+  void testQuerySingleGroup() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    var groupId = Utils.randomString(10);
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      var handler = new GroupHandler(admin);
+
+      try (var consumer =
+          Consumer.builder()
+              .groupId(groupId)
+              .topics(Set.of(topicName))
+              .bootstrapServers(bootstrapServers())
+              .build()) {
+        Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+        var group =
+            Assertions.assertInstanceOf(
+                GroupHandler.Group.class, handler.response(Optional.of(groupId), Map.of()));
+        Assertions.assertEquals(groupId, group.groupId);
+        Assertions.assertEquals(1, group.members.size());
+      }
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/web/HandlerTest.java b/app/src/test/java/org/astraea/web/HandlerTest.java
new file mode 100644
index 00000000..6e0420f2
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/HandlerTest.java
@@ -0,0 +1,46 @@
+package org.astraea.web;
+
+import com.sun.net.httpserver.HttpExchange;
+import java.net.URI;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class HandlerTest {
+
+  @Test
+  void testException() {
+    var exception = new IllegalStateException("hello");
+    Handler handler =
+        (paths, queries) -> {
+          throw exception;
+        };
+
+    var httpExchange = Mockito.mock(HttpExchange.class);
+    Mockito.when(httpExchange.getRequestURI()).thenReturn(URI.create("http://localhost:8888/abc"));
+    var r = Assertions.assertInstanceOf(ErrorObject.class, handler.response(httpExchange));
+    Assertions.assertNotEquals(200, r.code);
+    Assertions.assertEquals(exception.getMessage(), r.message);
+  }
+
+  @Test
+  void testParseTarget() {
+    Assertions.assertFalse(
+        Handler.parseTarget(URI.create("http://localhost:11111/abc")).isPresent());
+    var target = Handler.parseTarget(URI.create("http://localhost:11111/abc/bbb"));
+    Assertions.assertTrue(target.isPresent());
+    Assertions.assertEquals("bbb", target.get());
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> Handler.parseTarget(URI.create("http://localhost:11111/abc/bbb/dddd")));
+  }
+
+  @Test
+  void testParseQuery() {
+    var uri = URI.create("http://localhost:11111/abc?k=v&a=b");
+    var queries = Handler.parseQueries(uri);
+    Assertions.assertEquals(2, queries.size());
+    Assertions.assertEquals("v", queries.get("k"));
+    Assertions.assertEquals("b", queries.get("a"));
+  }
+}
diff --git a/app/src/test/java/org/astraea/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
new file mode 100644
index 00000000..3cf09750
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
@@ -0,0 +1,57 @@
+package org.astraea.web;
+
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import org.astraea.Utils;
+import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TopicHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testListTopics() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      admin.creator().topic(topicName).create();
+      TimeUnit.SECONDS.sleep(3);
+      var handler = new TopicHandler(admin);
+      var response =
+          Assertions.assertInstanceOf(
+              TopicHandler.Topics.class, handler.response(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());
+    }
+  }
+
+  @Test
+  void testQueryNonexistentTopic() {
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      var handler = new TopicHandler(admin);
+      var exception =
+          Assertions.assertThrows(
+              NoSuchElementException.class,
+              () -> handler.response(Optional.of("unknown"), Map.of()));
+      Assertions.assertTrue(exception.getMessage().contains("unknown"));
+    }
+  }
+
+  @Test
+  void testQuerySingleTopic() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+      admin.creator().topic(topicName).create();
+      TimeUnit.SECONDS.sleep(3);
+      var handler = new TopicHandler(admin);
+      var topicInfo =
+          Assertions.assertInstanceOf(
+              TopicHandler.TopicInfo.class, handler.response(Optional.of(topicName), Map.of()));
+      Assertions.assertEquals(topicName, topicInfo.name);
+      Assertions.assertNotEquals(0, topicInfo.configs.size());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/web/WebServiceTest.java b/app/src/test/java/org/astraea/web/WebServiceTest.java
new file mode 100644
index 00000000..403c4ee9
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/WebServiceTest.java
@@ -0,0 +1,17 @@
+package org.astraea.web;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class WebServiceTest {
+
+  @Test
+  void testArgument() {
+    var argument =
+        org.astraea.argument.Argument.parse(
+            new WebService.Argument(),
+            new String[] {"--bootstrap.servers", "localhost", "--port", "65535"});
+    Assertions.assertEquals("localhost", argument.bootstrapServers());
+    Assertions.assertEquals(65535, argument.port);
+  }
+}