Repository: skiptests/astraea
Pull request #416 (2022-06-21 17:42:56)

Implement post record web api
resolve https://github.com/skiptests/astraea/issues/397

chia7712: 另外這隻API的文件是要在另外一隻PR完成嗎？
brandboat: > 另外這隻API的文件是要在另外一隻PR完成嗎？

是的，連同 consumer web api 一同完成後，會另開 pr 撰寫 rest api /records 中文文件


Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/Admin.java b/app/src/main/java/org/astraea/app/admin/Admin.java
index 2038d3ef..e101b87c 100644
--- a/app/src/main/java/org/astraea/app/admin/Admin.java
+++ b/app/src/main/java/org/astraea/app/admin/Admin.java
@@ -216,6 +216,22 @@ public interface Admin extends Closeable {
    */
   Map<String, Transaction> transactions(Set<String> transactionIds);
 
+  /**
+   * remove an empty group. It causes error if the group has memebrs.
+   *
+   * @param groupId to remove
+   */
+  void removeGroup(String groupId);
+
+  /** @param groupId to remove all (dynamic and static) members */
+  void removeAllMembers(String groupId);
+
+  /**
+   * @param groupId to remove static members
+   * @param members group instance id (static member)
+   */
+  void removeStaticMembers(String groupId, Set<String> members);
+
   @Override
   void close();
 }
diff --git a/app/src/main/java/org/astraea/app/admin/Builder.java b/app/src/main/java/org/astraea/app/admin/Builder.java
index 27b91bfc..e09c6778 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -36,9 +36,11 @@ import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
 import org.apache.kafka.clients.admin.MemberDescription;
+import org.apache.kafka.clients.admin.MemberToRemove;
 import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.RemoveMembersFromConsumerGroupOptions;
 import org.apache.kafka.clients.admin.TransactionListing;
 import org.apache.kafka.common.ElectionType;
 import org.apache.kafka.common.TopicPartitionReplica;
@@ -551,6 +553,37 @@ public class Builder {
                   .collect(
                       Collectors.toMap(Map.Entry::getKey, e -> Transaction.from(e.getValue()))));
     }
+
+    @Override
+    public void removeGroup(String groupId) {
+      Utils.packException(() -> admin.deleteConsumerGroups(Set.of(groupId)).all().get());
+    }
+
+    @Override
+    public void removeAllMembers(String groupId) {
+      Utils.packException(
+          () ->
+              admin
+                  .removeMembersFromConsumerGroup(
+                      groupId, new RemoveMembersFromConsumerGroupOptions())
+                  .all()
+                  .get());
+    }
+
+    @Override
+    public void removeStaticMembers(String groupId, Set<String> members) {
+      Utils.packException(
+          () ->
+              admin
+                  .removeMembersFromConsumerGroup(
+                      groupId,
+                      new RemoveMembersFromConsumerGroupOptions(
+                          members.stream()
+                              .map(MemberToRemove::new)
+                              .collect(Collectors.toUnmodifiableList())))
+                  .all()
+                  .get());
+    }
   }
 
   private static class ConfigImpl implements Config {
diff --git a/app/src/main/java/org/astraea/app/consumer/Builder.java b/app/src/main/java/org/astraea/app/consumer/Builder.java
index d618f031..11a13084 100644
--- a/app/src/main/java/org/astraea/app/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/app/consumer/Builder.java
@@ -39,7 +39,8 @@ abstract class Builder<Key, Value> {
    * @return this builder
    */
   public Builder<Key, Value> fromBeginning() {
-    return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    this.configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    return this;
   }
 
   /**
@@ -48,7 +49,8 @@ abstract class Builder<Key, Value> {
    * @return this builder
    */
   public Builder<Key, Value> fromLatest() {
-    return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+    this.configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+    return this;
   }
 
   /**
@@ -76,23 +78,15 @@ abstract class Builder<Key, Value> {
     return (Builder<Key, NewValue>) this;
   }
 
-  public Builder<Key, Value> config(String key, String value) {
-    this.configs.put(key, value);
-    return this;
-  }
-
-  public Builder<Key, Value> configs(Map<String, String> configs) {
-    this.configs.putAll(configs);
-    return this;
-  }
-
   public Builder<Key, Value> bootstrapServers(String bootstrapServers) {
-    return config(
+    this.configs.put(
         ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(bootstrapServers));
+    return this;
   }
 
   public Builder<Key, Value> isolation(Isolation isolation) {
-    return config(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
+    this.configs.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
+    return this;
   }
 
   /** @return consumer instance. The different builders may return inherited consumer interface. */
diff --git a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
index 3f54e160..3c2d5e72 100644
--- a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
@@ -79,15 +79,13 @@ public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
     return (PartitionsBuilder<Key, NewValue>) super.valueDeserializer(valueDeserializer);
   }
 
-  @Override
   public PartitionsBuilder<Key, Value> config(String key, String value) {
-    super.config(key, value);
+    this.configs.put(key, value);
     return this;
   }
 
-  @Override
   public PartitionsBuilder<Key, Value> configs(Map<String, String> configs) {
-    super.configs(configs);
+    this.configs.putAll(configs);
     return this;
   }
 
diff --git a/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java b/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
index d2c05be9..3690f074 100644
--- a/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
+++ b/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
@@ -17,6 +17,7 @@
 package org.astraea.app.consumer;
 
 import java.time.Duration;
+import java.util.Optional;
 
 /**
  * This inherited consumer offers function related to consumer group.
@@ -32,4 +33,13 @@ public interface SubscribedConsumer<Key, Value> extends Consumer<Key, Value> {
    * @param timeout to wait commit.
    */
   void commitOffsets(Duration timeout);
+
+  /** @return the group id including this consumer */
+  String groupId();
+
+  /** @return the member id used by this consumer */
+  String memberId();
+
+  /** @return group instance id (static member) */
+  Optional<String> groupInstanceId();
 }
diff --git a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
index c975fba2..052c1e6e 100644
--- a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -21,6 +21,7 @@ import static java.util.Objects.requireNonNull;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -31,12 +32,14 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
 
   TopicsBuilder(Set<String> topics) {
     this.topics = requireNonNull(topics);
-    config(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis());
   }
 
   public TopicsBuilder<Key, Value> groupId(String groupId) {
-    config(ConsumerConfig.GROUP_ID_CONFIG, requireNonNull(groupId));
-    return this;
+    return config(ConsumerConfig.GROUP_ID_CONFIG, requireNonNull(groupId));
+  }
+
+  public TopicsBuilder<Key, Value> groupInstanceId(String groupInstanceId) {
+    return config(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, requireNonNull(groupInstanceId));
   }
 
   public TopicsBuilder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
@@ -91,15 +94,13 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     return (TopicsBuilder<Key, NewValue>) super.valueDeserializer(valueDeserializer);
   }
 
-  @Override
   public TopicsBuilder<Key, Value> config(String key, String value) {
-    super.config(key, value);
+    this.configs.put(key, value);
     return this;
   }
 
-  @Override
   public TopicsBuilder<Key, Value> configs(Map<String, String> configs) {
-    super.configs(configs);
+    this.configs.putAll(configs);
     return this;
   }
 
@@ -122,6 +123,9 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
   @SuppressWarnings("unchecked")
   @Override
   public SubscribedConsumer<Key, Value> build() {
+    // generate group id if it is empty
+    configs.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis());
+
     var kafkaConsumer =
         new KafkaConsumer<>(
             configs,
@@ -156,5 +160,19 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     public void commitOffsets(Duration timeout) {
       kafkaConsumer.commitSync(timeout);
     }
+
+    @Override
+    public String groupId() {
+      return kafkaConsumer.groupMetadata().groupId();
+    }
+
+    @Override
+    public String memberId() {
+      return kafkaConsumer.groupMetadata().memberId();
+    }
+
+    public Optional<String> groupInstanceId() {
+      return kafkaConsumer.groupMetadata().groupInstanceId();
+    }
   }
 }
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/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
new file mode 100644
index 00000000..331d64cf
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -0,0 +1,140 @@
+/*
+ * 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 JsonObject get(Optional<String> target, Map<String, String> queries) {
+    // TODO: Implement topic read web apis (https://github.com/skiptests/astraea/pull/405)
+    return ErrorObject.for404("GET is not supported yet");
+  }
+
+  @Override
+  public JsonObject 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) {
+      // TODO: Return HTTP status code 202 instead of 200
+      //  (https://github.com/skiptests/astraea/issues/420)
+      return new JsonObject() {};
+    }
+    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 JsonObject {
+    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/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/docker/start_app.sh b/docker/start_app.sh
index add5b565..7dbeb732 100755
--- a/docker/start_app.sh
+++ b/docker/start_app.sh
@@ -91,10 +91,12 @@ function runContainer() {
       # 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"
+        continue
       fi
       # this element must be port
       if [[ "$defined_port" == "true" ]]; then
         web_port="$word"
+        break
       fi
     done
     # manually add "--port" to make sure web service bind on random port we pass
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..20e5e535 100644
--- a/docs/web_server/web_api_groups_chinese.md
+++ b/docs/web_server/web_api_groups_chinese.md
@@ -1,6 +1,9 @@
 /groups
 ===
 
+Consumer groups 是 Kafka 提供給讀取端的負載平衡機制，該機制允許我們建立多個 consumer instances，它們會透過 Kafka broker 來互相溝通，
+並且協調各自該負責讀取哪些 partitions，藉此達到平衡以及容錯等功能。 Groups APIs 可以幫助我們觀察 groups 內各個 consumer member 的狀態
+
 - [查詢所有 groups](#查詢所有-groups)
 - [查詢指定 group](#查詢指定-group)
 
@@ -9,6 +12,12 @@
 GET /groups
 ```
 
+參數
+
+| 名稱                       | 說明                                   | 預設  |
+|--------------------------|--------------------------------------|-----|
+| topic                    | (選填) 只查詢跟此 topic 有關的 consumer groups | 無   |
+
 cURL 範例
 ```shell
 curl -X GET http://localhost:8001/groups
@@ -59,6 +68,12 @@ JSON Response 範例
 GET /groups/{groupId}
 ```
 
+參數
+
+| 名稱                       | 說明                                   | 預設  |
+|--------------------------|--------------------------------------|-----|
+| topic                    | (選填) 只查詢跟此 topic 有關的 consumer groups | 無   |
+
 cURL 範例
 查詢名為 group-1 的 group 資訊
 ```shell
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/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index 1430fdd6..d5bd7d91 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -34,6 +34,7 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
@@ -155,11 +156,6 @@ public class AdminTest extends RequireBrokerCluster {
                 Assertions.assertEquals(0, offset.earliest());
                 Assertions.assertEquals(0, offset.latest());
               });
-
-      admin.creator().topic("a").numberOfPartitions(3).create();
-      // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(3);
-      Assertions.assertEquals(6, admin.offsets().size());
     }
   }
 
@@ -835,4 +831,102 @@ public class AdminTest extends RequireBrokerCluster {
           transaction.topicPartitions().size());
     }
   }
+
+  @Test
+  void testRemoveAllMembers() {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
+        var consumer =
+            Consumer.forTopics(Set.of(topicName))
+                .bootstrapServers(bootstrapServers())
+                .fromBeginning()
+                .build()) {
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      Assertions.assertEquals(1, consumer.poll(1, Duration.ofSeconds(5)).size());
+
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      admin.removeAllMembers(consumer.groupId());
+      Assertions.assertEquals(
+          0,
+          admin
+              .consumerGroups(Set.of(consumer.groupId()))
+              .get(consumer.groupId())
+              .activeMembers()
+              .size());
+    }
+  }
+
+  @Test
+  void testRemoveStaticMembers() {
+    var topicName = Utils.randomString(10);
+    var staticId = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
+        var consumer =
+            Consumer.forTopics(Set.of(topicName))
+                .bootstrapServers(bootstrapServers())
+                .groupInstanceId(staticId)
+                .fromBeginning()
+                .build()) {
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      Assertions.assertEquals(1, consumer.poll(1, Duration.ofSeconds(5)).size());
+
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      admin.removeStaticMembers(consumer.groupId(), Set.of(consumer.groupInstanceId().get()));
+      Assertions.assertEquals(
+          0,
+          admin
+              .consumerGroups(Set.of(consumer.groupId()))
+              .get(consumer.groupId())
+              .activeMembers()
+              .size());
+    }
+  }
+
+  @Test
+  void testRemoveGroupWithDynamicMembers() {
+    var groupId = Utils.randomString(10);
+    var topicName = Utils.randomString(10);
+    try (var consumer =
+        Consumer.forTopics(Set.of(topicName))
+            .bootstrapServers(bootstrapServers())
+            .groupId(groupId)
+            .build()) {
+      Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+    }
+    try (var admin = Admin.of(bootstrapServers())) {
+      Assertions.assertTrue(admin.consumerGroupIds().contains(groupId));
+      admin.removeGroup(groupId);
+      Assertions.assertFalse(admin.consumerGroupIds().contains(groupId));
+    }
+  }
+
+  @Test
+  void testRemoveGroupWithStaticMembers() throws InterruptedException, ExecutionException {
+    var groupId = Utils.randomString(10);
+    var topicName = Utils.randomString(10);
+    try (var consumer =
+        Consumer.forTopics(Set.of(topicName))
+            .bootstrapServers(bootstrapServers())
+            .groupId(groupId)
+            .groupInstanceId(Utils.randomString(10))
+            .build()) {
+      Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+    }
+
+    try (var admin = Admin.of(bootstrapServers())) {
+      Assertions.assertTrue(admin.consumerGroupIds().contains(groupId));
+      // the static member is existent
+      Assertions.assertThrows(GroupNotEmptyException.class, () -> admin.removeGroup(groupId));
+      // cleanup members
+      admin.removeAllMembers(groupId);
+      admin.removeGroup(groupId);
+      Assertions.assertFalse(admin.consumerGroupIds().contains(groupId));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
index d44ba18f..82b74bbf 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -112,8 +112,8 @@ public class ConsumerTest extends RequireBrokerCluster {
 
   @Test
   void testGroupId() {
-    var groupId = "testGroupId";
-    var topic = "testGroupId";
+    var groupId = Utils.randomString(10);
+    var topic = Utils.randomString(10);
     produceData(topic, 1);
 
     java.util.function.BiConsumer<String, Integer> testConsumer =
@@ -126,6 +126,9 @@ public class ConsumerTest extends RequireBrokerCluster {
                   .build()) {
             Assertions.assertEquals(
                 expectedSize, consumer.poll(expectedSize, Duration.ofSeconds(5)).size());
+            Assertions.assertEquals(id, consumer.groupId());
+            Assertions.assertNotNull(consumer.memberId());
+            Assertions.assertFalse(consumer.groupInstanceId().isPresent());
           }
         };
 
@@ -138,6 +141,19 @@ public class ConsumerTest extends RequireBrokerCluster {
     testConsumer.accept("another_group", 0);
   }
 
+  @Test
+  void testGroupInstanceId() {
+    var staticId = Utils.randomString(10);
+    try (var consumer =
+        Consumer.forTopics(Set.of(Utils.randomString(10)))
+            .bootstrapServers(bootstrapServers())
+            .groupInstanceId(staticId)
+            .build()) {
+      Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(2)).size());
+      Assertions.assertEquals(staticId, consumer.groupInstanceId().get());
+    }
+  }
+
   @Test
   void testDistanceFromLatest() {
     var count = 10;
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..9ff84589
--- /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(
+            JsonObject.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(new JsonObject() {}.json(), result.json());
+
+    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)));
+  }
+}