Repository: skiptests/astraea
Pull request #415 (2022-06-21 06:44:32)

Allow `ClusterLogAllocation#migrationReplica` specify data directory
延續 https://github.com/skiptests/astraea/pull/412#issuecomment-1159626602

令 `ClusterLogAllocation#migrationReplica` 能夠指定要搬移到哪個 data directory，同時修改 `ShufflePlanGenerator` 的實作，使他能夠使用這個新的 API 來做產生計劃，這樣產生的計劃比較明確，能使後續 `CostFunction` 比較好做預測。


Patch:
diff --git a/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java b/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
index 6c7c4e82..0e1f0443 100644
--- a/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
+++ b/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
@@ -17,13 +17,12 @@
 package org.astraea.app.balancer.generator;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.RebalancePlanProposal;
@@ -47,8 +46,6 @@ import org.astraea.app.cost.NodeInfo;
  *       LogManager#nextLogDirs method implementation in Apache Kafka server for more details.
  *   <li>Change the leader/follower of a partition by a member of this replica set, the original
  *       leader/follower becomes a follower/leader.
- *   <li>Change the data directory of one replica from current data directory to another on the
- *       current broker.
  * </ol>
  */
 public class ShufflePlanGenerator implements RebalancePlanGenerator {
@@ -71,33 +68,11 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
     return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
   }
 
-  private int migrationSelector(List<Movement> movementCandidates) {
-    // There are two kinds of Movement, each operating on different kind of candidate.
-    // * The Movement#replicaSetMovement selecting target by the broker ID
-    // * The Movement#dataDirectoryMovement selecting target by the data dir on the specific broker
-    // If we mix all these movements and randomly picking one as the selected migration, doing so
-    // will cause fairness issue. Given a concrete example, there are 100 brokers, and each broker
-    // have 3 data directories. Now the leader log of a topic/partition with 3 replicas is being
-    // selected. How many possible migrations over here? The Answer is:
-    // 1. 99 from Movement#replicaSetMigration since the leader log can move to any other broker.
-    // 2. 2 from Movement#dataDirectoryMigration since the leader log can move to the other 2 dirs.
-    // So there are 101 possible migrations. But these "two ways" of migration have different
-    // probability to occur. We can expect a large number of replica set changes occur but only a
-    // few for the data directory migration. This will make the ShufflePlanGenerator hard to address
-    // specific kind of balance issue due to <strong>it rarely propose them</strong>.
-    final var migrationTypeList =
-        IntStream.range(0, movementCandidates.size())
-            .mapToObj(i -> Map.entry(movementCandidates.get(i).getClass(), i))
-            .collect(
-                Collectors.groupingBy(
-                    Map.Entry::getKey,
-                    Collectors.mapping(Map.Entry::getValue, Collectors.toUnmodifiableList())))
-            .values()
-            .stream()
-            .collect(Collectors.toUnmodifiableList());
-    final var selectedMigration =
-        migrationTypeList.get(ThreadLocalRandom.current().nextInt(0, migrationTypeList.size()));
-    return selectedMigration.get(ThreadLocalRandom.current().nextInt(0, selectedMigration.size()));
+  private <T> T randomElement(Collection<T> collection) {
+    return collection.stream()
+        .skip(ThreadLocalRandom.current().nextInt(0, collection.size()))
+        .findFirst()
+        .orElseThrow();
   }
 
   @Override
@@ -147,14 +122,17 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
 
             Consumer<Integer> replicaSetMigration =
                 (targetBroker) -> {
-                  newAllocation.migrateReplica(sourceTopicPartition, sourceBroker, targetBroker);
+                  var destDir = randomElement(clusterInfo.dataDirectories(targetBroker));
+                  newAllocation.migrateReplica(
+                      sourceTopicPartition, sourceBroker, targetBroker, destDir);
                   rebalancePlanBuilder.addInfo(
                       String.format(
-                          "Change replica set of topic %s partition %d, from %d to %d.",
+                          "Change replica set of topic %s partition %d, from %d to %d at %s.",
                           sourceTopicPartition.topic(),
                           sourceTopicPartition.partition(),
                           sourceLogPlacement.broker(),
-                          targetBroker));
+                          targetBroker,
+                          destDir));
                 };
             Consumer<LogPlacement> leaderFollowerMigration =
                 (newLeaderReplicaCandidate) -> {
@@ -169,19 +147,6 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
                           sourceIsLeader ? "leader" : "follower",
                           sourceIsLeader ? "follower" : "leader"));
                 };
-            Consumer<String> dataDirectoryMigration =
-                (dataDirectory) -> {
-                  newAllocation.changeDataDirectory(
-                      sourceTopicPartition, sourceBroker, dataDirectory);
-                  rebalancePlanBuilder.addInfo(
-                      String.format(
-                          "Change the data directory of topic %s partition %d replica at broker %d, from %s to %s",
-                          sourceTopicPartition.topic(),
-                          sourceTopicPartition.partition(),
-                          sourceLogPlacement.broker(),
-                          sourceLogPlacement.logDirectory().map(Object::toString).orElse("unknown"),
-                          dataDirectory));
-                };
 
             // generate a set of valid migration broker for given placement.
             final var validMigrationCandidates = new ArrayList<Movement>();
@@ -210,16 +175,10 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
                         () -> leaderFollowerMigration.accept(sourceLogPlacement)));
               }
             }
-            // [Valid movement 3] change the data directory of selected replica
-            clusterInfo.dataDirectories(sourceLogPlacement.broker()).stream()
-                .filter(dir -> !dir.equals(sourceLogPlacement.logDirectory().orElse(null)))
-                .map(dir -> (Runnable) () -> dataDirectoryMigration.accept(dir))
-                .map(Movement::dataDirectoryMovement)
-                .forEach(validMigrationCandidates::add);
 
             // pick a migration and execute
-            final var selectedMigrationIndex = migrationSelector(validMigrationCandidates);
-            validMigrationCandidates.get(selectedMigrationIndex).run();
+            final var selectedMigrationIndex = randomElement(validMigrationCandidates);
+            selectedMigrationIndex.run();
           }
 
           return rebalancePlanBuilder.withRebalancePlan(newAllocation).build();
@@ -231,13 +190,7 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
     static ReplicaSetMovement replicaSetMovement(Runnable runnable) {
       return runnable::run;
     }
-
-    static DataDirectoryMovement dataDirectoryMovement(Runnable runnable) {
-      return runnable::run;
-    }
   }
 
   interface ReplicaSetMovement extends Movement {}
-
-  interface DataDirectoryMovement extends Movement {}
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index 48c63dd8..0daff1eb 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -27,15 +27,35 @@ import org.astraea.app.admin.TopicPartition;
  */
 public interface ClusterLogAllocation {
 
-  /** let specific broker leave the replica set and let another broker join the replica set. */
-  void migrateReplica(TopicPartition topicPartition, int atBroker, int toBroker);
+  /**
+   * let specific broker leave the replica set and let another broker join the replica set. Which
+   * data directory the migrated replica will be is up to the Kafka broker implementation to decide.
+   *
+   * @param topicPartition the topic/partition to perform replica migration
+   * @param atBroker the id of the broker about to remove
+   * @param toBroker the id of the broker about to replace the removed broker
+   */
+  default void migrateReplica(TopicPartition topicPartition, int atBroker, int toBroker) {
+    migrateReplica(topicPartition, atBroker, toBroker, null);
+  }
+  // TODO: Revise the log argument by TopicPartitionReplica, once #411 is merged
+
+  /**
+   * let specific broker leave the replica set and let another broker join the replica set.
+   *
+   * @param topicPartition the topic/partition to perform replica migration
+   * @param atBroker the id of the broker about to remove
+   * @param toBroker the id of the broker about to replace the removed broker
+   * @param toDir the absolute path of the data directory this migrated replica is supposed to be on
+   *     the destination broker, if {@code null} is specified then the data directory choice is left
+   *     up to the Kafka broker implementation.
+   */
+  void migrateReplica(TopicPartition topicPartition, int atBroker, int toBroker, String toDir);
+  // TODO: Revise the log argument by TopicPartitionReplica, once #411 is merged
 
   /** let specific follower log become the leader log of this topic/partition. */
   void letReplicaBecomeLeader(TopicPartition topicPartition, int followerReplica);
 
-  /** change the data directory of specific log */
-  void changeDataDirectory(TopicPartition topicPartition, int atBroker, String newPath);
-
   /** Retrieve the log placements of specific {@link TopicPartition}. */
   List<LogPlacement> logPlacements(TopicPartition topicPartition);
 
diff --git a/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java
index 386b57ab..29eaf8d2 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java
@@ -163,7 +163,7 @@ public class LayeredClusterLogAllocation implements ClusterLogAllocation {
 
   @Override
   public synchronized void migrateReplica(
-      TopicPartition topicPartition, int broker, int destinationBroker) {
+      TopicPartition topicPartition, int broker, int destinationBroker, String toDir) {
     ensureNotLocked();
 
     final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
@@ -176,18 +176,13 @@ public class LayeredClusterLogAllocation implements ClusterLogAllocation {
       throw new IllegalMigrationException(
           broker + " is not part of the replica set for " + topicPartition);
 
-    int destinationLogIndex = indexOfBroker(sourceLogPlacements, destinationBroker).orElse(-1);
-    if (destinationLogIndex != -1)
-      throw new IllegalArgumentException(
-          destinationBroker + " is already part of the replica set, no need to move");
-
     this.allocation.put(
         topicPartition,
         IntStream.range(0, sourceLogPlacements.size())
             .mapToObj(
                 index ->
                     index == sourceLogIndex
-                        ? LogPlacement.of(destinationBroker)
+                        ? LogPlacement.of(destinationBroker, toDir)
                         : sourceLogPlacements.get(index))
             .collect(Collectors.toUnmodifiableList()));
   }
@@ -227,30 +222,6 @@ public class LayeredClusterLogAllocation implements ClusterLogAllocation {
             .collect(Collectors.toUnmodifiableList()));
   }
 
-  @Override
-  public synchronized void changeDataDirectory(
-      TopicPartition topicPartition, int broker, String path) {
-    ensureNotLocked();
-
-    final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
-    if (sourceLogPlacements == null)
-      throw new IllegalMigrationException(
-          topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
-
-    int sourceLogIndex = indexOfBroker(sourceLogPlacements, broker).orElse(-1);
-    if (sourceLogIndex == -1)
-      throw new IllegalMigrationException(
-          broker + " is not part of the replica set for " + topicPartition);
-
-    final var oldLog = this.logPlacements(topicPartition).get(sourceLogIndex);
-    final var newLog = LogPlacement.of(oldLog.broker(), path);
-    this.allocation.put(
-        topicPartition,
-        IntStream.range(0, sourceLogPlacements.size())
-            .mapToObj(index -> index == sourceLogIndex ? newLog : sourceLogPlacements.get(index))
-            .collect(Collectors.toUnmodifiableList()));
-  }
-
   @Override
   public List<LogPlacement> logPlacements(TopicPartition topicPartition) {
     if (allocation.containsKey(topicPartition)) return allocation.get(topicPartition);
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 2909de73..7e74b82a 100644
--- a/app/src/main/java/org/astraea/app/web/BeanHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BeanHandler.java
@@ -37,7 +37,7 @@ public class BeanHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> domain, Map<String, String> properties) {
+  public Response get(Optional<String> domain, Map<String, String> properties) {
     var builder = BeanQuery.builder().usePropertyListPattern().properties(properties);
     domain.ifPresent(builder::domainName);
     return new NodeBeans(
@@ -52,7 +52,7 @@ public class BeanHandler implements Handler {
             .collect(Collectors.toUnmodifiableList()));
   }
 
-  static class Property implements JsonObject {
+  static class Property implements Response {
     final String key;
     final String value;
 
@@ -62,7 +62,7 @@ public class BeanHandler implements Handler {
     }
   }
 
-  static class Attribute implements JsonObject {
+  static class Attribute implements Response {
     final String key;
     final String value;
 
@@ -72,7 +72,7 @@ public class BeanHandler implements Handler {
     }
   }
 
-  static class Bean implements JsonObject {
+  static class Bean implements Response {
     final String domainName;
     final List<Property> properties;
     final List<Attribute> attributes;
@@ -90,7 +90,7 @@ public class BeanHandler implements Handler {
     }
   }
 
-  static class NodeBean implements JsonObject {
+  static class NodeBean implements Response {
     final String host;
     final List<Bean> beans;
 
@@ -100,7 +100,7 @@ public class BeanHandler implements Handler {
     }
   }
 
-  static class NodeBeans implements JsonObject {
+  static class NodeBeans implements Response {
     final List<NodeBean> nodeBeans;
 
     NodeBeans(List<NodeBean> nodeBeans) {
diff --git a/app/src/main/java/org/astraea/app/web/BrokerHandler.java b/app/src/main/java/org/astraea/app/web/BrokerHandler.java
index 25a47838..a7bbfa27 100644
--- a/app/src/main/java/org/astraea/app/web/BrokerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BrokerHandler.java
@@ -44,7 +44,7 @@ class BrokerHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+  public Response get(Optional<String> target, Map<String, String> queries) {
 
     var brokers =
         admin.brokers(brokers(target)).entrySet().stream()
@@ -54,7 +54,7 @@ class BrokerHandler implements Handler {
     return new Brokers(brokers);
   }
 
-  static class Topic implements JsonObject {
+  static class Topic implements Response {
     final String topic;
     final int partitionCount;
 
@@ -64,7 +64,7 @@ class BrokerHandler implements Handler {
     }
   }
 
-  static class Broker implements JsonObject {
+  static class Broker implements Response {
     final int id;
     final List<Topic> topics;
     final Map<String, String> configs;
@@ -84,7 +84,7 @@ class BrokerHandler implements Handler {
     }
   }
 
-  static class Brokers implements JsonObject {
+  static class Brokers implements Response {
     final List<Broker> brokers;
 
     Brokers(List<Broker> brokers) {
diff --git a/app/src/main/java/org/astraea/app/web/GroupHandler.java b/app/src/main/java/org/astraea/app/web/GroupHandler.java
index a2f015b9..ebc1f1e4 100644
--- a/app/src/main/java/org/astraea/app/web/GroupHandler.java
+++ b/app/src/main/java/org/astraea/app/web/GroupHandler.java
@@ -26,6 +26,7 @@ import org.astraea.app.admin.TopicPartition;
 
 public class GroupHandler implements Handler {
   static final String TOPIC_KEY = "topic";
+  static final String INSTANCE_KEY = "instance";
   private final Admin admin;
 
   GroupHandler(Admin admin) {
@@ -37,7 +38,21 @@ public class GroupHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+  public Response delete(String groupId, Map<String, String> queries) {
+    var groupInstanceId = queries.get(INSTANCE_KEY);
+    var activeMembers = admin.consumerGroups(Set.of(groupId)).get(groupId).activeMembers();
+    // Deleting all members can't work when there is no members already.
+    if (groupInstanceId == null && !activeMembers.isEmpty()) admin.removeAllMembers(groupId);
+    // Deleting nonexistent instance id can cause error
+    if (groupInstanceId != null
+        && activeMembers.stream()
+            .anyMatch(m -> m.groupInstanceId().filter(g -> g.equals(groupInstanceId)).isPresent()))
+      admin.removeStaticMembers(groupId, Set.of(groupInstanceId));
+    return Response.OK;
+  }
+
+  @Override
+  public Response get(Optional<String> target, Map<String, String> queries) {
     var topics =
         queries.containsKey(TOPIC_KEY) ? Set.of(queries.get(TOPIC_KEY)) : admin.topicNames();
     var consumerGroups = admin.consumerGroups(groupIds(target));
@@ -96,7 +111,7 @@ public class GroupHandler implements Handler {
     return new Groups(groups);
   }
 
-  static class OffsetProgress implements JsonObject {
+  static class OffsetProgress implements Response {
     final String topic;
     final int partitionId;
     final long earliest;
@@ -112,7 +127,7 @@ public class GroupHandler implements Handler {
     }
   }
 
-  static class Member implements JsonObject {
+  static class Member implements Response {
     final String memberId;
     final String groupInstanceId;
     final String clientId;
@@ -133,7 +148,7 @@ public class GroupHandler implements Handler {
     }
   }
 
-  static class Group implements JsonObject {
+  static class Group implements Response {
     final String groupId;
     final List<Member> members;
 
@@ -143,7 +158,7 @@ public class GroupHandler implements Handler {
     }
   }
 
-  static class Groups implements JsonObject {
+  static class Groups implements Response {
     final List<Group> groups;
 
     Groups(List<Group> groups) {
diff --git a/app/src/main/java/org/astraea/app/web/Handler.java b/app/src/main/java/org/astraea/app/web/Handler.java
index 07e3fe19..982dd9f3 100644
--- a/app/src/main/java/org/astraea/app/web/Handler.java
+++ b/app/src/main/java/org/astraea/app/web/Handler.java
@@ -37,10 +37,9 @@ interface Handler extends HttpHandler {
     return target.map(Set::of).orElse(all);
   }
 
-  default JsonObject process(HttpExchange exchange) {
+  default Response process(HttpExchange exchange) {
     var method = exchange.getRequestMethod().toUpperCase(Locale.ROOT);
     var start = System.currentTimeMillis();
-    JsonObject response;
     try {
       switch (method) {
         case "GET":
@@ -49,13 +48,14 @@ interface Handler extends HttpHandler {
           return post(PostRequest.of(exchange));
         case "DELETE":
           var target = parseTarget(exchange.getRequestURI());
-          if (target.isPresent()) return delete(target.get());
+          if (target.isPresent())
+            return delete(target.get(), parseQueries(exchange.getRequestURI()));
         default:
-          return ErrorObject.for404(method + " is not supported yet");
+          return Response.NOT_FOUND;
       }
     } catch (Exception e) {
       e.printStackTrace();
-      return new ErrorObject(e);
+      return Response.of(e);
     } finally {
       System.out.println(
           "take "
@@ -67,14 +67,12 @@ interface Handler extends HttpHandler {
 
   @Override
   default void handle(HttpExchange exchange) throws IOException {
-    JsonObject response = process(exchange);
-
+    Response response = process(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);
+    exchange.sendResponseHeaders(response.code(), responseData.length);
     try (var os = exchange.getResponseBody()) {
       os.write(responseData);
     }
@@ -106,7 +104,7 @@ interface Handler extends HttpHandler {
    * @param queries queries from url
    * @return json object to return
    */
-  JsonObject get(Optional<String> target, Map<String, String> queries);
+  Response get(Optional<String> target, Map<String, String> queries);
 
   /**
    * handle the post request.
@@ -114,8 +112,8 @@ interface Handler extends HttpHandler {
    * @param request the last keyword of url
    * @return json object to return
    */
-  default JsonObject post(PostRequest request) {
-    return ErrorObject.for404("POST is not supported yet");
+  default Response post(PostRequest request) {
+    return Response.NOT_FOUND;
   }
 
   /**
@@ -124,7 +122,7 @@ interface Handler extends HttpHandler {
    * @param target the last keyword of url
    * @return json object to return
    */
-  default JsonObject delete(String target) {
-    return ErrorObject.for404("DELETE is not supported yet");
+  default Response delete(String target, Map<String, String> queries) {
+    return Response.NOT_FOUND;
   }
 }
diff --git a/app/src/main/java/org/astraea/app/web/JsonObject.java b/app/src/main/java/org/astraea/app/web/JsonObject.java
deleted file mode 100644
index 7bdb4c16..00000000
--- a/app/src/main/java/org/astraea/app/web/JsonObject.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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 com.google.gson.Gson;
-
-interface JsonObject {
-  default String json() {
-    return new Gson().toJson(this);
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/web/PipelineHandler.java b/app/src/main/java/org/astraea/app/web/PipelineHandler.java
index 82d5c99d..191bdabf 100644
--- a/app/src/main/java/org/astraea/app/web/PipelineHandler.java
+++ b/app/src/main/java/org/astraea/app/web/PipelineHandler.java
@@ -39,7 +39,7 @@ class PipelineHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+  public Response get(Optional<String> target, Map<String, String> queries) {
     var tps =
         topicPartitions(admin).stream()
             .filter(filter(queries))
@@ -85,7 +85,7 @@ class PipelineHandler implements Handler {
         .collect(Collectors.toUnmodifiableList());
   }
 
-  static class Producer implements JsonObject {
+  static class Producer implements Response {
     final long producerId;
     final int producerEpoch;
     final int lastSequence;
@@ -99,7 +99,7 @@ class PipelineHandler implements Handler {
     }
   }
 
-  static class Consumer implements JsonObject {
+  static class Consumer implements Response {
     final String groupId;
     final String memberId;
     final String groupInstanceId;
@@ -115,7 +115,7 @@ class PipelineHandler implements Handler {
     }
   }
 
-  static class TopicPartition implements JsonObject {
+  static class TopicPartition implements Response {
     final String topic;
     final int partition;
     final Collection<Producer> from = new ArrayList<>();
@@ -127,7 +127,7 @@ class PipelineHandler implements Handler {
     }
   }
 
-  static class TopicPartitions implements JsonObject {
+  static class TopicPartitions implements Response {
     final Collection<TopicPartition> topicPartitions;
 
     TopicPartitions(Collection<TopicPartition> topicPartitions) {
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 8b4ec77e..05f75299 100644
--- a/app/src/main/java/org/astraea/app/web/PostRequest.java
+++ b/app/src/main/java/org/astraea/app/web/PostRequest.java
@@ -89,4 +89,12 @@ public interface PostRequest {
   default short shortValue(String key) {
     return Short.parseShort(value(key));
   }
+
+  default boolean booleanValue(String key, boolean defaultValue) {
+    return get(key).map(Boolean::parseBoolean).orElse(defaultValue);
+  }
+
+  default boolean booleanValue(String key) {
+    return Boolean.parseBoolean(value(key));
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/web/ProducerHandler.java b/app/src/main/java/org/astraea/app/web/ProducerHandler.java
index 44868119..90aa3a8c 100644
--- a/app/src/main/java/org/astraea/app/web/ProducerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/ProducerHandler.java
@@ -23,7 +23,6 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.ProducerState;
 import org.astraea.app.admin.TopicPartition;
 
 class ProducerHandler implements Handler {
@@ -57,7 +56,7 @@ class ProducerHandler implements Handler {
     return new Partitions(topics);
   }
 
-  static class ProducerState implements JsonObject {
+  static class ProducerState implements Response {
 
     final long producerId;
     final int producerEpoch;
@@ -72,7 +71,7 @@ class ProducerHandler implements Handler {
     }
   }
 
-  static class Partition implements JsonObject {
+  static class Partition implements Response {
     final String topic;
     final int partition;
     final List<ProducerState> states;
@@ -87,7 +86,7 @@ class ProducerHandler implements Handler {
     }
   }
 
-  static class Partitions implements JsonObject {
+  static class Partitions implements Response {
     final List<Partition> partitions;
 
     Partitions(List<Partition> partitions) {
diff --git a/app/src/main/java/org/astraea/app/web/QuotaHandler.java b/app/src/main/java/org/astraea/app/web/QuotaHandler.java
index 2eb17e7e..0413e920 100644
--- a/app/src/main/java/org/astraea/app/web/QuotaHandler.java
+++ b/app/src/main/java/org/astraea/app/web/QuotaHandler.java
@@ -51,7 +51,7 @@ public class QuotaHandler implements Handler {
   }
 
   @Override
-  public JsonObject post(PostRequest request) {
+  public Response post(PostRequest request) {
     if (request.get(IP_KEY).isPresent()) {
       admin
           .quotaCreator()
@@ -70,10 +70,10 @@ public class QuotaHandler implements Handler {
       return new Quotas(
           admin.quotas(org.astraea.app.admin.Quota.Target.CLIENT_ID, request.value(CLIENT_ID_KEY)));
     }
-    return ErrorObject.for404("You must define either " + CLIENT_ID_KEY + " or " + IP_KEY);
+    return Response.NOT_FOUND;
   }
 
-  static class Target implements JsonObject {
+  static class Target implements Response {
     final String name;
     final String value;
 
@@ -83,7 +83,7 @@ public class QuotaHandler implements Handler {
     }
   }
 
-  static class Limit implements JsonObject {
+  static class Limit implements Response {
     final String name;
     final double value;
 
@@ -93,7 +93,7 @@ public class QuotaHandler implements Handler {
     }
   }
 
-  static class Quota implements JsonObject {
+  static class Quota implements Response {
     final Target target;
     final Limit limit;
 
@@ -111,7 +111,7 @@ public class QuotaHandler implements Handler {
     }
   }
 
-  static class Quotas implements JsonObject {
+  static class Quotas implements Response {
     final List<Quota> quotas;
 
     Quotas(Collection<org.astraea.app.admin.Quota> quotas) {
diff --git a/app/src/main/java/org/astraea/app/web/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
new file mode 100644
index 00000000..163bbe92
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -0,0 +1,137 @@
+/*
+ * 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 static java.util.Objects.requireNonNull;
+
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.BiFunction;
+import org.astraea.app.common.Utils;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.producer.Serializer;
+
+public class RecordHandler implements Handler {
+  static final String KEY_SERIALIZER = "keySerializer";
+  static final String VALUE_SERIALIZER = "valueSerializer";
+  static final String TOPIC = "topic";
+  static final String KEY = "key";
+  static final String VALUE = "value";
+  static final String TIMESTAMP = "timestamp";
+  static final String PARTITION = "partition";
+  static final String ASYNC = "async";
+
+  // visible for testing
+  final Producer<byte[], byte[]> producer;
+
+  RecordHandler(String bootstrapServers) {
+    this.producer = Producer.builder().bootstrapServers(requireNonNull(bootstrapServers)).build();
+  }
+
+  @Override
+  public Response get(Optional<String> target, Map<String, String> queries) {
+    // TODO: Implement topic read web apis (https://github.com/skiptests/astraea/pull/405)
+    return Response.NOT_FOUND;
+  }
+
+  @Override
+  public Response post(PostRequest request) {
+    var topic =
+        request.get(TOPIC).orElseThrow(() -> new IllegalArgumentException("topic must be set"));
+    var keySerDe =
+        SerDe.valueOf(
+            request.get(KEY_SERIALIZER).map(String::toUpperCase).orElse(SerDe.STRING.name()));
+    var valueSerDe =
+        SerDe.valueOf(
+            request.get(VALUE_SERIALIZER).map(String::toUpperCase).orElse(SerDe.STRING.name()));
+    var async = request.booleanValue(ASYNC, false);
+    var sender = producer.sender().topic(topic);
+
+    // TODO: Support headers (https://github.com/skiptests/astraea/issues/422)
+    request.get(KEY).ifPresent(k -> sender.key(keySerDe.serializer.apply(topic, k)));
+    request.get(VALUE).ifPresent(v -> sender.value(valueSerDe.serializer.apply(topic, v)));
+    request.get(TIMESTAMP).ifPresent(t -> sender.timestamp(Long.parseLong(t)));
+    request.get(PARTITION).ifPresent(p -> sender.partition(Integer.parseInt(p)));
+
+    var senderFuture = sender.run().toCompletableFuture();
+
+    if (async) return Response.ACCEPT;
+
+    return Utils.packException(() -> new Metadata(senderFuture.get()));
+  }
+
+  enum SerDe {
+    BYTEARRAY(
+        (topic, value) ->
+            Optional.ofNullable(value).map(v -> Base64.getDecoder().decode(v)).orElse(null)),
+    STRING(
+        (topic, value) ->
+            Optional.ofNullable(value)
+                .map(v -> Serializer.STRING.serialize(topic, List.of(), value))
+                .orElse(null)),
+    LONG(
+        (topic, value) ->
+            Optional.ofNullable(value)
+                .map(Long::parseLong)
+                .map(longVal -> Serializer.LONG.serialize(topic, List.of(), longVal))
+                .orElse(null)),
+    INTEGER(
+        (topic, value) ->
+            Optional.ofNullable(value)
+                .map(Integer::parseInt)
+                .map(intVal -> Serializer.INTEGER.serialize(topic, List.of(), intVal))
+                .orElse(null)),
+    FLOAT(
+        (topic, value) ->
+            Optional.ofNullable(value)
+                .map(Float::parseFloat)
+                .map(floatVal -> Serializer.FLOAT.serialize(topic, List.of(), floatVal))
+                .orElse(null)),
+    DOUBLE(
+        (topic, value) ->
+            Optional.ofNullable(value)
+                .map(Double::parseDouble)
+                .map(doubleVal -> Serializer.DOUBLE.serialize(topic, List.of(), doubleVal))
+                .orElse(null));
+
+    final BiFunction<String, String, byte[]> serializer;
+
+    SerDe(BiFunction<String, String, byte[]> serializer) {
+      this.serializer = requireNonNull(serializer);
+    }
+  }
+
+  static class Metadata implements Response {
+    final String topic;
+    final int partition;
+    final long offset;
+    final long timestamp;
+    final int serializedKeySize;
+    final int serializedValueSize;
+
+    Metadata(org.astraea.app.producer.Metadata metadata) {
+      topic = metadata.topic();
+      partition = metadata.partition();
+      offset = metadata.offset();
+      timestamp = metadata.timestamp();
+      serializedKeySize = metadata.serializedKeySize();
+      serializedValueSize = metadata.serializedValueSize();
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/web/ErrorObject.java b/app/src/main/java/org/astraea/app/web/Response.java
similarity index 53%
rename from app/src/main/java/org/astraea/app/web/ErrorObject.java
rename to app/src/main/java/org/astraea/app/web/Response.java
index c96e83e7..d483c448 100644
--- a/app/src/main/java/org/astraea/app/web/ErrorObject.java
+++ b/app/src/main/java/org/astraea/app/web/Response.java
@@ -16,24 +16,21 @@
  */
 package org.astraea.app.web;
 
+import com.google.gson.Gson;
 import java.util.NoSuchElementException;
 
-class ErrorObject implements JsonObject {
+interface Response {
 
-  static ErrorObject for404(String message) {
-    return new ErrorObject(404, message);
-  }
-
-  final int code;
-  final String message;
+  Response OK = new ResponseImpl(200);
+  Response ACCEPT = new ResponseImpl(202);
+  Response NOT_FOUND = new ResponseImpl(404);
 
-  ErrorObject(Exception exception) {
-    this(code(exception), exception.getMessage());
+  static Response of(Exception exception) {
+    return new ResponseImpl(code(exception), exception.getMessage());
   }
 
-  ErrorObject(int code, String message) {
-    this.code = code;
-    this.message = message;
+  static Response for404(String message) {
+    return new ResponseImpl(404, message);
   }
 
   private static int code(Exception exception) {
@@ -41,4 +38,37 @@ class ErrorObject implements JsonObject {
     if (exception instanceof NoSuchElementException) return 404;
     return 400;
   }
+
+  /** @return http code */
+  default int code() {
+    return 200;
+  }
+
+  default String json() {
+    return new Gson().toJson(this);
+  }
+
+  class ResponseImpl implements Response {
+    final int code;
+    final String message;
+
+    ResponseImpl(int code) {
+      this(code, "");
+    }
+
+    ResponseImpl(int code, String message) {
+      this.code = code;
+      this.message = message;
+    }
+
+    @Override
+    public int code() {
+      return code;
+    }
+
+    @Override
+    public String json() {
+      return message == null || message.isEmpty() ? "" : new Gson().toJson(this);
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/web/TopicHandler.java b/app/src/main/java/org/astraea/app/web/TopicHandler.java
index 0429855c..468a397e 100644
--- a/app/src/main/java/org/astraea/app/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/app/web/TopicHandler.java
@@ -45,11 +45,11 @@ class TopicHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+  public Response get(Optional<String> target, Map<String, String> queries) {
     return get(topicNames(target));
   }
 
-  private JsonObject get(Set<String> topicNames) {
+  private Response get(Set<String> topicNames) {
     var topics = admin.topics(topicNames);
     var replicas = admin.replicas(topics.keySet());
     var partitions =
@@ -86,7 +86,7 @@ class TopicHandler implements Handler {
   }
 
   @Override
-  public JsonObject post(PostRequest request) {
+  public Response post(PostRequest request) {
     admin
         .creator()
         .topic(request.value(TOPIC_NAME_KEY))
@@ -106,7 +106,7 @@ class TopicHandler implements Handler {
     return new TopicInfo(request.value(TOPIC_NAME_KEY), List.of(), Map.of());
   }
 
-  static class Topics implements JsonObject {
+  static class Topics implements Response {
     final Collection<TopicInfo> topics;
 
     private Topics(Collection<TopicInfo> topics) {
@@ -114,7 +114,7 @@ class TopicHandler implements Handler {
     }
   }
 
-  static class TopicInfo implements JsonObject {
+  static class TopicInfo implements Response {
     final String name;
     final List<Partition> partitions;
     final Map<String, String> configs;
@@ -134,7 +134,7 @@ class TopicHandler implements Handler {
     }
   }
 
-  static class Partition implements JsonObject {
+  static class Partition implements Response {
     final int id;
     final long earliest;
     final long latest;
@@ -148,7 +148,7 @@ class TopicHandler implements Handler {
     }
   }
 
-  static class Replica implements JsonObject {
+  static class Replica implements Response {
     final int broker;
     final long lag;
     final long size;
diff --git a/app/src/main/java/org/astraea/app/web/TransactionHandler.java b/app/src/main/java/org/astraea/app/web/TransactionHandler.java
index d6bbd5a2..68413a53 100644
--- a/app/src/main/java/org/astraea/app/web/TransactionHandler.java
+++ b/app/src/main/java/org/astraea/app/web/TransactionHandler.java
@@ -32,7 +32,7 @@ class TransactionHandler implements Handler {
   }
 
   @Override
-  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+  public Response get(Optional<String> target, Map<String, String> queries) {
     var transactions =
         admin.transactions(Handler.compare(admin.transactionIds(), target)).entrySet().stream()
             .map(e -> new Transaction(e.getKey(), e.getValue()))
@@ -41,7 +41,7 @@ class TransactionHandler implements Handler {
     return new Transactions(transactions);
   }
 
-  static class TopicPartition implements JsonObject {
+  static class TopicPartition implements Response {
     final String topic;
     final int partition;
 
@@ -51,7 +51,7 @@ class TransactionHandler implements Handler {
     }
   }
 
-  static class Transaction implements JsonObject {
+  static class Transaction implements Response {
     final String id;
     final int coordinatorId;
     final String state;
@@ -74,7 +74,7 @@ class TransactionHandler implements Handler {
     }
   }
 
-  static class Transactions implements JsonObject {
+  static class Transactions implements Response {
     final List<Transaction> transactions;
 
     Transactions(List<Transaction> transactions) {
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 f43cf392..4deeff96 100644
--- a/app/src/main/java/org/astraea/app/web/WebService.java
+++ b/app/src/main/java/org/astraea/app/web/WebService.java
@@ -45,6 +45,7 @@ public class WebService {
     server.createContext("/transactions", new TransactionHandler(Admin.of(arg.configs())));
     if (arg.needJmx())
       server.createContext("/beans", new BeanHandler(Admin.of(arg.configs()), arg.jmxPorts()));
+    server.createContext("/records", new RecordHandler(arg.bootstrapServers()));
     server.start();
   }
 
diff --git a/docs/web_server/README.md b/docs/web_server/README.md
index 331fb21e..55b508f5 100644
--- a/docs/web_server/README.md
+++ b/docs/web_server/README.md
@@ -11,6 +11,7 @@ Astraea 建立了一套 Web Server 服務，使用者可以透過簡易好上手
 
 - `--bootstrap.servers`: broker 位址與連接埠，用以索取叢集資訊，可填寫多台 brokers，並以 `,` 做區隔
 - `--port`: web server 連接埠，此參數為選填值，不填寫的話會選一個連接埠隨機綁定
+- `--jmx.port`: 目標叢集各節點所使用的JMX port，這個值指定後就可以使用 `beans` APIs
 
 ## Web APIs
 - [/topics](./web_api_topics_chinese.md)
@@ -18,3 +19,6 @@ Astraea 建立了一套 Web Server 服務，使用者可以透過簡易好上手
 - [/brokers](./web_api_brokers_chinese.md)
 - [/producers](./web_api_producers_chinese.md)
 - [/quotas](./web_api_quotas_chinese.md)
+- [/transactions](./web_api_transactions_chinese.md)
+- [/pipelines](./web_api_pipelines_chinese.md)
+- [/beans](./web_api_beans_chinese.md)
diff --git a/docs/web_server/web_api_beans_chinese.md b/docs/web_server/web_api_beans_chinese.md
new file mode 100644
index 00000000..375278b7
--- /dev/null
+++ b/docs/web_server/web_api_beans_chinese.md
@@ -0,0 +1,70 @@
+/beans
+===
+
+Beans 可以用來顯示伺服器端所記錄的 Java Metrics
+
+- [查詢所有 beans](#查詢所有節點的-beans)
+
+## 查詢所有節點的 beans
+```shell
+GET /beans/{domain}?{property_key=property_value}
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/beans/kafka.cluster?partition=22&name=UnderMinIsr
+```
+
+參數
+
+| 名稱             | 說明                                        | 預設  |
+|----------------|-------------------------------------------|-----|
+| domain         | (選填) 只搜尋擁有此 domain 的 Java Metrics         | 無   |
+| property_key   | (選填) 只搜尋擁有此 property_key 的 Java Metrics   | 無   |
+| property_value | (選填) 只搜尋擁有此 property_value 的 Java Metrics | 無   |
+
+JSON Response 範例
+- `nodeBeans`
+  - `host`: 節點名稱
+  - `beans`:
+    - `domainName`: domain of object name
+    - `properties`: property of object name，可以與 domain 組成 object name 用來定位 Java Metrics
+    - `attributes`: Java Metrics 所記錄的值，同樣是 key=value 的格式
+```json
+{
+  "nodeBeans": [
+    {
+      "host": "192.168.50.168",
+      "beans": [
+        {
+          "domainName": "kafka.cluster",
+          "properties": [
+            {
+              "key": "partition",
+              "value": "22"
+            },
+            {
+              "key": "name",
+              "value": "UnderMinIsr"
+            },
+            {
+              "key": "topic",
+              "value": "__consumer_offsets"
+            },
+            {
+              "key": "type",
+              "value": "Partition"
+            }
+          ],
+          "attributes": [
+            {
+              "key": "Value",
+              "value": "0"
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/web_server/web_api_groups_chinese.md b/docs/web_server/web_api_groups_chinese.md
index 7b92c6c6..604ba969 100644
--- a/docs/web_server/web_api_groups_chinese.md
+++ b/docs/web_server/web_api_groups_chinese.md
@@ -1,14 +1,24 @@
 /groups
 ===
 
+Consumer groups 是 Kafka 提供給讀取端的負載平衡機制，該機制允許我們建立多個 consumer instances，它們會透過 Kafka broker 來互相溝通，
+並且協調各自該負責讀取哪些 partitions，藉此達到平衡以及容錯等功能。 Groups APIs 可以幫助我們觀察 groups 內各個 consumer member 的狀態
+
 - [查詢所有 groups](#查詢所有-groups)
 - [查詢指定 group](#查詢指定-group)
+- [清除 members](#清除-members)
 
 ## 查詢所有 groups
 ```shell
 GET /groups
 ```
 
+參數
+
+| 名稱                       | 說明                                   | 預設  |
+|--------------------------|--------------------------------------|-----|
+| topic                    | (選填) 只查詢跟此 topic 有關的 consumer groups | 無   |
+
 cURL 範例
 ```shell
 curl -X GET http://localhost:8001/groups
@@ -59,6 +69,12 @@ JSON Response 範例
 GET /groups/{groupId}
 ```
 
+參數
+
+| 名稱                       | 說明                                   | 預設  |
+|--------------------------|--------------------------------------|-----|
+| topic                    | (選填) 只查詢跟此 topic 有關的 consumer groups | 無   |
+
 cURL 範例
 查詢名為 group-1 的 group 資訊
 ```shell
@@ -87,3 +103,21 @@ JSON Response 範例
   ]
 }
 ```
+
+## 清除 members
+```shell
+DELETE /groups/{groupId}?{instance=id}
+```
+
+參數
+
+| 名稱                       | 說明                                             | 預設  |
+|--------------------------|------------------------------------------------|-----|
+| groupId                    | (必填) 清除此group底下所有的consumer members             | 無   |
+| instance                    | (選填) 只清除此group instance id 關聯的 consumer member | 無   |
+
+cURL 範例
+刪除 group-1 底下所有的 consumer members
+```shell
+curl -X DELETE http://localhost:8001/groups/group-1
+```
diff --git a/docs/web_server/web_api_pipelines_chinese.md b/docs/web_server/web_api_pipelines_chinese.md
new file mode 100644
index 00000000..4d5c01f7
--- /dev/null
+++ b/docs/web_server/web_api_pipelines_chinese.md
@@ -0,0 +1,69 @@
+/pipelines
+===
+
+Pipeline 是用來描述一個 kafka partition 的資料來源以及資料目的，它根據 Kafka 所追蹤的 consumer group 和
+idempotent/transaction producer 的資訊來拼湊出每個 partitions 是被哪些 consumers/producers 所使用
+
+- [查詢所有 pipelines](#查詢所有-pipelines)
+
+## 查詢所有 pipelines
+```shell
+GET /pipelines
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/pipelines
+```
+
+參數
+
+| 名稱                       | 說明                                        | 預設    |
+|--------------------------|-------------------------------------------|-------|
+| active                   | (選填) 是否只顯示活躍的 partitions (也就是 from/to 有值) | false |
+
+JSON Response 範例
+- `topicPartitions`
+  - `topic`: topic 名稱
+  - `partition`: partition id
+  - `from`: 有哪些 producers 正在寫資料到這個 partition ，注意，只有啟動 idempotent/transaction producer 才會被kafka所紀錄
+  - `to`: 有哪些 consumers 正在讀取資料，注意，只有使用 consumer group 的 consumers 才會被kafka所紀錄
+```json
+{
+  "topicPartitions": [
+    {
+      "topic": "testPerformance-1656075398393",
+      "partition": 0,
+      "from": [
+        {
+          "producerId": 0,
+          "producerEpoch": 0,
+          "lastSequence": 64354,
+          "lastTimestamp": 1656075698698
+        }
+      ],
+      "to": []
+    },
+    {
+      "topic": "testPerformance-1656075901856",
+      "partition": 0,
+      "from": [
+        {
+          "producerId": 1,
+          "producerEpoch": 0,
+          "lastSequence": 2244,
+          "lastTimestamp": 1656075909846
+        }
+      ],
+      "to": [
+        {
+          "groupId": "groupId-1656075902062",
+          "memberId": "consumer-groupId-1656075902062-1-b39f51a9-8a2a-46d5-aa5d-8595a014978f",
+          "clientId": "consumer-groupId-1656075902062-1",
+          "host": "/172.17.0.1"
+        }
+      ]
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/web_server/web_api_producers_chinese.md b/docs/web_server/web_api_producers_chinese.md
index 31f9e39d..61f9f2d5 100644
--- a/docs/web_server/web_api_producers_chinese.md
+++ b/docs/web_server/web_api_producers_chinese.md
@@ -1,6 +1,8 @@
 /producers
 ===
 
+producers APIs 可以幫助我們查看啟動 idempotent/transaction 行為的 producers
+
 - [查詢 producers](#查詢-producers)
 
 ## 查詢 producers
diff --git a/docs/web_server/web_api_transactions_chinese.md b/docs/web_server/web_api_transactions_chinese.md
new file mode 100644
index 00000000..aa4ef048
--- /dev/null
+++ b/docs/web_server/web_api_transactions_chinese.md
@@ -0,0 +1,70 @@
+/transactions
+===
+
+- [查詢所有 transactions](#查詢所有-transactions)
+- [查詢指定 transaction](#查詢指定-transaction)
+
+## 查詢所有 transactions
+```shell
+GET /transactions
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/transactions
+```
+
+JSON Response 範例
+- `id`: transaction id
+- `coordinatorId`: 負責協調此次transaction行為的節點id (也就是broker id)
+- `states`: 目前交易的狀態
+- `producerId`: producer id，從 0 開始遞增
+- `producerEpoch`: producer epoch，從 0 開始遞增，用以保證 producer transaction 的唯一性
+- `transactionTimeoutMs`: 此次交易可持續多久
+- `topicPartitions`: 此次交易涉及的partitions
+```json
+{
+  "transactions": [
+    {
+      "id": "abc",
+      "coordinatorId": 1001,
+      "state": "Ongoing",
+      "producerId": 0,
+      "producerEpoch": 0,
+      "transactionTimeoutMs": 60000,
+      "topicPartitions": [
+        {
+          "topic": "testPerformance-1656075398393",
+          "partition": 0
+        }
+      ]
+    }
+  ]
+}
+```
+
+## 查詢指定 transaction
+
+```shell
+GET /transactions/{transaction id}
+```
+
+cURL 範例
+
+查詢 abc 的交易行為
+```shell
+curl -X GET http://localhost:8001/transactions/abc
+```
+
+JSON Response 範例
+ ```json
+{
+  "id": "abc",
+  "coordinatorId": 1001,
+  "state": "CompleteCommit",
+  "producerId": 0,
+  "producerEpoch": 0,
+  "transactionTimeoutMs": 60000,
+  "topicPartitions": []
+}
+ ```


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java b/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
index 270ecc22..233f671c 100644
--- a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
@@ -25,6 +25,8 @@ import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.cost.ClusterInfoProvider;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 class LayeredClusterLogAllocationTest {
 
@@ -60,41 +62,65 @@ class LayeredClusterLogAllocationTest {
 
     Assertions.assertEquals(
         1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
+    Assertions.assertNull(
+        clusterLogAllocation
+            .logPlacements(sourceTopicPartition)
+            .get(0)
+            .logDirectory()
+            .orElse(null));
     Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
   }
 
-  @Test
-  void letReplicaBecomeLeader() {
+  @ParameterizedTest
+  @ValueSource(
+      strings = {"null", "/tmp/data-directory-0", "/tmp/data-directory-1", "/tmp/data-directory-2"})
+  void migrateReplica(String dataDirectory) {
+    dataDirectory = dataDirectory.equals("null") ? null : dataDirectory;
     final var fakeClusterInfo =
-        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 2, (i) -> Set.of("topic"));
+        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 1, (i) -> Set.of("topic"));
     final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
-    final var sourceTopicPartition = TopicPartition.of("topic", "0");
+    final var sourceTopicPartition0 = TopicPartition.of("topic", "0");
 
-    clusterLogAllocation.letReplicaBecomeLeader(sourceTopicPartition, 1);
+    clusterLogAllocation.migrateReplica(sourceTopicPartition0, 0, 1, dataDirectory);
 
     Assertions.assertEquals(
-        1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
+        1, clusterLogAllocation.logPlacements(sourceTopicPartition0).get(0).broker());
     Assertions.assertEquals(
-        0, clusterLogAllocation.logPlacements(sourceTopicPartition).get(1).broker());
+        dataDirectory,
+        clusterLogAllocation
+            .logPlacements(sourceTopicPartition0)
+            .get(0)
+            .logDirectory()
+            .orElse(null));
+
+    final var sourceTopicPartition1 = TopicPartition.of("topic", "0");
+    clusterLogAllocation.migrateReplica(sourceTopicPartition1, 1, 1, dataDirectory);
+    Assertions.assertEquals(
+        1, clusterLogAllocation.logPlacements(sourceTopicPartition1).get(0).broker());
+    Assertions.assertEquals(
+        dataDirectory,
+        clusterLogAllocation
+            .logPlacements(sourceTopicPartition1)
+            .get(0)
+            .logDirectory()
+            .orElse(null));
+
     Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
   }
 
   @Test
-  void changeDataDirectory() {
+  void letReplicaBecomeLeader() {
     final var fakeClusterInfo =
-        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 1, (i) -> Set.of("topic"));
+        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 2, (i) -> Set.of("topic"));
     final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
     final var sourceTopicPartition = TopicPartition.of("topic", "0");
 
-    clusterLogAllocation.changeDataDirectory(sourceTopicPartition, 0, "/path/to/somewhere");
+    clusterLogAllocation.letReplicaBecomeLeader(sourceTopicPartition, 1);
 
     Assertions.assertEquals(
-        "/path/to/somewhere",
-        clusterLogAllocation
-            .logPlacements(sourceTopicPartition)
-            .get(0)
-            .logDirectory()
-            .orElseThrow());
+        1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
+    Assertions.assertEquals(
+        0, clusterLogAllocation.logPlacements(sourceTopicPartition).get(1).broker());
     Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
   }
 
@@ -127,10 +153,8 @@ class LayeredClusterLogAllocationTest {
         .flatMap(x -> fakeClusterInfo.replicas(x).stream())
         .forEach(
             replica ->
-                allocation1.changeDataDirectory(
-                    TopicPartition.of(replica.topic(), Integer.toString(replica.partition())),
-                    0,
-                    "/yeah"));
+                allocation1.letReplicaBecomeLeader(
+                    TopicPartition.of(replica.topic(), Integer.toString(replica.partition())), 1));
 
     final var allTopicPartitions =
         allocation1
@@ -160,8 +184,8 @@ class LayeredClusterLogAllocationTest {
 
     // can modify before lock
     Assertions.assertDoesNotThrow(() -> allocation.migrateReplica(toModify, 0, 9));
-    Assertions.assertDoesNotThrow(() -> allocation.letReplicaBecomeLeader(toModify, 1));
-    Assertions.assertDoesNotThrow(() -> allocation.changeDataDirectory(toModify, 2, "/nowhere"));
+    Assertions.assertDoesNotThrow(() -> allocation.migrateReplica(toModify, 1, 8, "somewhere"));
+    Assertions.assertDoesNotThrow(() -> allocation.letReplicaBecomeLeader(toModify, 2));
 
     final var extended = LayeredClusterLogAllocation.of(allocation);
 
@@ -169,14 +193,13 @@ class LayeredClusterLogAllocationTest {
     Assertions.assertThrows(
         IllegalStateException.class, () -> allocation.migrateReplica(toModify, 9, 0));
     Assertions.assertThrows(
-        IllegalStateException.class, () -> allocation.letReplicaBecomeLeader(toModify, 0));
+        IllegalStateException.class, () -> allocation.migrateReplica(toModify, 0, 9, "somewhere"));
     Assertions.assertThrows(
-        IllegalStateException.class,
-        () -> allocation.changeDataDirectory(toModify, 2, "/anywhere"));
+        IllegalStateException.class, () -> allocation.letReplicaBecomeLeader(toModify, 0));
 
     // the extended one can modify
     Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 9, 0));
+    Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 8, 1, "somewhere"));
     Assertions.assertDoesNotThrow(() -> extended.letReplicaBecomeLeader(toModify, 0));
-    Assertions.assertDoesNotThrow(() -> extended.changeDataDirectory(toModify, 2, "/anywhere"));
   }
 }
diff --git a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
index cddc63b6..578c914e 100644
--- a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
@@ -50,15 +50,9 @@ public class GroupHandlerTest extends RequireBrokerCluster {
         var response =
             Assertions.assertInstanceOf(
                 GroupHandler.Groups.class, handler.get(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());
-        response
-            .groups
-            .iterator()
-            .next()
-            .members
-            .forEach(m -> Assertions.assertNull(m.groupInstanceId));
+        var group = response.groups.stream().filter(g -> g.groupId.equals(groupId)).findAny().get();
+        Assertions.assertEquals(1, group.members.size());
+        group.members.forEach(m -> Assertions.assertNull(m.groupInstanceId));
       }
     }
   }
@@ -157,4 +151,69 @@ public class GroupHandlerTest extends RequireBrokerCluster {
       }
     }
   }
+
+  @Test
+  void testDeleteMembers() {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new GroupHandler(admin);
+
+      // test 0: delete all members
+      try (var consumer =
+          Consumer.forTopics(Set.of(topicName)).bootstrapServers(bootstrapServers()).build()) {
+        Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+        Assertions.assertEquals(
+            1,
+            admin
+                .consumerGroups(Set.of(consumer.groupId()))
+                .get(consumer.groupId())
+                .activeMembers()
+                .size());
+
+        handler.delete(consumer.groupId(), Map.of());
+        Assertions.assertEquals(
+            0,
+            admin
+                .consumerGroups(Set.of(consumer.groupId()))
+                .get(consumer.groupId())
+                .activeMembers()
+                .size());
+
+        // idempotent test
+        handler.delete(consumer.groupId(), Map.of());
+      }
+
+      // test 1: delete static member
+      try (var consumer =
+          Consumer.forTopics(Set.of(topicName))
+              .bootstrapServers(bootstrapServers())
+              .groupInstanceId(Utils.randomString(10))
+              .build()) {
+        Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+        Assertions.assertEquals(
+            1,
+            admin
+                .consumerGroups(Set.of(consumer.groupId()))
+                .get(consumer.groupId())
+                .activeMembers()
+                .size());
+
+        handler.delete(
+            consumer.groupId(),
+            Map.of(GroupHandler.INSTANCE_KEY, consumer.groupInstanceId().get()));
+        Assertions.assertEquals(
+            0,
+            admin
+                .consumerGroups(Set.of(consumer.groupId()))
+                .get(consumer.groupId())
+                .activeMembers()
+                .size());
+
+        // idempotent test
+        handler.delete(
+            consumer.groupId(),
+            Map.of(GroupHandler.INSTANCE_KEY, consumer.groupInstanceId().get()));
+      }
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/HandlerTest.java b/app/src/test/java/org/astraea/app/web/HandlerTest.java
index 6941e8b9..8b54598c 100644
--- a/app/src/test/java/org/astraea/app/web/HandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/HandlerTest.java
@@ -16,8 +16,13 @@
  */
 package org.astraea.app.web;
 
+import com.sun.net.httpserver.Headers;
 import com.sun.net.httpserver.HttpExchange;
+import java.io.IOException;
+import java.io.OutputStream;
 import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -35,7 +40,7 @@ public class HandlerTest {
     var exchange = Mockito.mock(HttpExchange.class);
     Mockito.when(exchange.getRequestURI()).thenReturn(URI.create("http://localhost:8888/abc"));
     Mockito.when(exchange.getRequestMethod()).thenReturn("get");
-    var r = Assertions.assertInstanceOf(ErrorObject.class, handler.process(exchange));
+    var r = Assertions.assertInstanceOf(Response.ResponseImpl.class, handler.process(exchange));
     Assertions.assertNotEquals(200, r.code);
     Assertions.assertEquals(exception.getMessage(), r.message);
   }
@@ -60,4 +65,25 @@ public class HandlerTest {
     Assertions.assertEquals("v", queries.get("k"));
     Assertions.assertEquals("b", queries.get("a"));
   }
+
+  @Test
+  void testNoResponseBody() throws IOException {
+    Handler handler =
+        new Handler() {
+          @Override
+          public Response get(Optional<String> target, Map<String, String> queries) {
+            return null;
+          }
+
+          @Override
+          public Response process(HttpExchange exchange) {
+            return Response.OK;
+          }
+        };
+    var he = Mockito.mock(HttpExchange.class);
+    Mockito.when(he.getResponseHeaders()).thenReturn(Mockito.mock(Headers.class));
+    Mockito.when(he.getResponseBody()).thenReturn(Mockito.mock(OutputStream.class));
+    handler.handle(he);
+    Mockito.verify(he).sendResponseHeaders(200, 0);
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
new file mode 100644
index 00000000..69ce157f
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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 static java.nio.charset.StandardCharsets.UTF_8;
+import static org.astraea.app.web.RecordHandler.ASYNC;
+import static org.astraea.app.web.RecordHandler.KEY;
+import static org.astraea.app.web.RecordHandler.KEY_SERIALIZER;
+import static org.astraea.app.web.RecordHandler.PARTITION;
+import static org.astraea.app.web.RecordHandler.TIMESTAMP;
+import static org.astraea.app.web.RecordHandler.TOPIC;
+import static org.astraea.app.web.RecordHandler.VALUE;
+import static org.astraea.app.web.RecordHandler.VALUE_SERIALIZER;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Base64;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Stream;
+import org.astraea.app.common.Utils;
+import org.astraea.app.consumer.Consumer;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.astraea.app.web.RecordHandler.Metadata;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class RecordHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testInvalidPost() {
+    var handler = new RecordHandler(bootstrapServers());
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> handler.post(PostRequest.of(Map.of())),
+        "topic must be set");
+  }
+
+  @Test
+  void testPost() {
+    var topic = Utils.randomString(10);
+    var handler = new RecordHandler(bootstrapServers());
+    var currentTimestamp = System.currentTimeMillis();
+    var result =
+        Assertions.assertInstanceOf(
+            Metadata.class,
+            handler.post(
+                PostRequest.of(
+                    Map.of(
+                        KEY_SERIALIZER, "string",
+                        VALUE_SERIALIZER, "integer",
+                        TOPIC, topic,
+                        KEY, "foo",
+                        VALUE, "100",
+                        TIMESTAMP, "" + currentTimestamp,
+                        PARTITION, "0"))));
+
+    Assertions.assertEquals(0, result.offset);
+    Assertions.assertEquals(0, result.partition);
+    Assertions.assertEquals(topic, result.topic);
+    Assertions.assertEquals("foo".getBytes(UTF_8).length, result.serializedKeySize);
+    Assertions.assertEquals(4, result.serializedValueSize);
+
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .build()) {
+      var record = consumer.poll(1, Duration.ofSeconds(10)).iterator().next();
+      Assertions.assertEquals(topic, record.topic());
+      Assertions.assertEquals(currentTimestamp, record.timestamp());
+      Assertions.assertEquals(0, record.partition());
+      Assertions.assertEquals("foo".getBytes(UTF_8).length, record.serializedKeySize());
+      Assertions.assertEquals(4, record.serializedValueSize());
+      Assertions.assertArrayEquals("foo".getBytes(UTF_8), record.key());
+      Assertions.assertArrayEquals(ByteBuffer.allocate(4).putInt(100).array(), record.value());
+    }
+  }
+
+  @Test
+  void testPostWithAsync() {
+    var topic = Utils.randomString(10);
+    var handler = new RecordHandler(bootstrapServers());
+    var currentTimestamp = System.currentTimeMillis();
+    var result =
+        Assertions.assertInstanceOf(
+            Response.class,
+            handler.post(
+                PostRequest.of(
+                    Map.of(
+                        KEY_SERIALIZER, "string",
+                        VALUE_SERIALIZER, "integer",
+                        TOPIC, topic,
+                        KEY, "foo",
+                        VALUE, "100",
+                        TIMESTAMP, "" + currentTimestamp,
+                        ASYNC, "true",
+                        PARTITION, "0"))));
+    Assertions.assertEquals(Response.ACCEPT, result);
+
+    handler.producer.flush();
+
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .build()) {
+      var record = consumer.poll(1, Duration.ofSeconds(10)).iterator().next();
+      Assertions.assertEquals(topic, record.topic());
+      Assertions.assertEquals(currentTimestamp, record.timestamp());
+      Assertions.assertEquals(0, record.partition());
+      Assertions.assertEquals("foo".getBytes(UTF_8).length, record.serializedKeySize());
+      Assertions.assertEquals(4, record.serializedValueSize());
+      Assertions.assertArrayEquals("foo".getBytes(UTF_8), record.key());
+      Assertions.assertArrayEquals(ByteBuffer.allocate(4).putInt(100).array(), record.value());
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("forTestSerializer")
+  void testSerializer(String serializer, String actual, byte[] expected) {
+    var topic = Utils.randomString(10);
+    var handler = new RecordHandler(bootstrapServers());
+    Assertions.assertInstanceOf(
+        Metadata.class,
+        handler.post(
+            PostRequest.of(Map.of(KEY_SERIALIZER, serializer, TOPIC, topic, KEY, actual))));
+
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .build()) {
+      var record = consumer.poll(1, Duration.ofSeconds(10)).iterator().next();
+      Assertions.assertArrayEquals(expected, record.key());
+    }
+  }
+
+  private static Stream<Arguments> forTestSerializer() {
+    return Stream.of(
+        arguments("integer", "10", ByteBuffer.allocate(Integer.BYTES).putInt(10).array()),
+        arguments("long", "11", ByteBuffer.allocate(Long.BYTES).putLong(11).array()),
+        arguments("float", "0.1", ByteBuffer.allocate(Float.BYTES).putFloat(0.1f).array()),
+        arguments("double", "0.1", ByteBuffer.allocate(Double.BYTES).putDouble(0.1).array()),
+        arguments("string", "astraea", "astraea".getBytes(UTF_8)),
+        arguments(
+            "bytearray",
+            Base64.getEncoder().encodeToString("astraea".getBytes(UTF_8)),
+            "astraea".getBytes(UTF_8)));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/web/ErrorObjectTest.java b/app/src/test/java/org/astraea/app/web/ResponseTest.java
similarity index 53%
rename from app/src/test/java/org/astraea/app/web/ErrorObjectTest.java
rename to app/src/test/java/org/astraea/app/web/ResponseTest.java
index 1b4d821a..0796e4c3 100644
--- a/app/src/test/java/org/astraea/app/web/ErrorObjectTest.java
+++ b/app/src/test/java/org/astraea/app/web/ResponseTest.java
@@ -16,15 +16,30 @@
  */
 package org.astraea.app.web;
 
+import java.nio.charset.StandardCharsets;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class ErrorObjectTest {
+public class ResponseTest {
 
   @Test
-  void test404() {
-    var error = ErrorObject.for404("this");
-    Assertions.assertEquals(404, error.code);
-    Assertions.assertEquals("this", error.message);
+  void testNotFound() {
+    Assertions.assertEquals(0, Response.NOT_FOUND.json().length());
+    Assertions.assertEquals(404, Response.NOT_FOUND.code());
+    Assertions.assertEquals(0, Response.NOT_FOUND.json().getBytes(StandardCharsets.UTF_8).length);
+  }
+
+  @Test
+  void testOk() {
+    Assertions.assertEquals(0, Response.OK.json().length());
+    Assertions.assertEquals(200, Response.OK.code());
+    Assertions.assertEquals(0, Response.OK.json().getBytes(StandardCharsets.UTF_8).length);
+  }
+
+  @Test
+  void testAccept() {
+    Assertions.assertEquals(0, Response.ACCEPT.json().length());
+    Assertions.assertEquals(202, Response.ACCEPT.code());
+    Assertions.assertEquals(0, Response.ACCEPT.json().getBytes(StandardCharsets.UTF_8).length);
   }
 }