Repository: skiptests/astraea
Pull request #464 (2022-07-09 11:33:40)

Support transactionId and multi records in POST /records
resolve https://github.com/skiptests/astraea/issues/432


Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java b/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java
index 01d33afe..5a7fd2dd 100644
--- a/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java
+++ b/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java
@@ -27,7 +27,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -184,13 +183,9 @@ public class ReplicaSyncingMonitor {
             .forEach(tpr -> previousCheckedSize.remove(tpr.getKey()));
       }
 
-      Utils.packException(
-          () -> {
-            long expectedWaitNs = argument.interval.toNanos();
-            long elapsedNs = (System.nanoTime() - startTime);
-            TimeUnit.NANOSECONDS.sleep(Math.max(expectedWaitNs - elapsedNs, 0));
-            return 0;
-          });
+      long expectedWaitNs = argument.interval.toNanos();
+      long elapsedNs = (System.nanoTime() - startTime);
+      Utils.sleep(Duration.ofNanos(Math.max(expectedWaitNs - elapsedNs, 0)));
     }
   }
 
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
index ff4815a2..6d4126e9 100644
--- a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
@@ -21,7 +21,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -176,7 +175,7 @@ class RebalanceAdminImpl implements RebalanceAdmin {
                     .map(x -> x.inSync() && !x.isFuture())
                     .orElse(false);
             // debounce & retrial interval
-            Utils.packException(() -> TimeUnit.MILLISECONDS.sleep(retrialTime.get().toMillis()));
+            Utils.sleep(retrialTime.get());
             debounce = synced ? (debounce - 1) : 2;
             // synced
             if (synced && debounce <= 0) return true;
diff --git a/app/src/main/java/org/astraea/app/common/Utils.java b/app/src/main/java/org/astraea/app/common/Utils.java
index 4180000a..22d9af55 100644
--- a/app/src/main/java/org/astraea/app/common/Utils.java
+++ b/app/src/main/java/org/astraea/app/common/Utils.java
@@ -21,8 +21,11 @@ import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.nio.file.Files;
 import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
 import java.util.SortedMap;
 import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -152,7 +155,7 @@ public final class Utils {
       try {
         var r = supplier.get();
         if (r != null) return r;
-        TimeUnit.SECONDS.sleep(1);
+        Utils.sleep(Duration.ofSeconds(1));
       } catch (Exception e) {
         lastError = e;
       }
@@ -177,6 +180,11 @@ public final class Utils {
     return (lastTime + interval.toMillis()) < System.currentTimeMillis();
   }
 
+  /**
+   * Perform a sleep using the duration. InterruptedException is wrapped to RuntimeException.
+   *
+   * @param duration to sleep
+   */
   public static void sleep(Duration duration) {
     Utils.swallowException(() -> TimeUnit.MILLISECONDS.sleep(duration.toMillis()));
   }
@@ -226,5 +234,10 @@ public final class Utils {
         TreeMap::new);
   }
 
+  public static <T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<T>> futures) {
+    return CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0]))
+        .thenApply(f -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));
+  }
+
   private Utils() {}
 }
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 426940b6..6dd9ca56 100644
--- a/app/src/main/java/org/astraea/app/web/PostRequest.java
+++ b/app/src/main/java/org/astraea/app/web/PostRequest.java
@@ -17,6 +17,7 @@
 package org.astraea.app.web;
 
 import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
 import com.google.gson.reflect.TypeToken;
 import com.sun.net.httpserver.HttpExchange;
 import java.io.IOException;
@@ -50,6 +51,12 @@ public interface PostRequest {
       var value = (double) obj;
       if (value - Math.floor(value) == 0) return String.valueOf((long) Math.floor(value));
     }
+    // TODO: handle double in nested type
+    // use gson instead of obj.toString in nested type since obj.toString won't add double quote to
+    // string and will create invalid json
+    if (obj instanceof Map || obj instanceof List) {
+      return new GsonBuilder().disableHtmlEscaping().create().toJson(obj);
+    }
     return obj.toString();
   }
 
@@ -72,6 +79,15 @@ public interface PostRequest {
     return Optional.ofNullable(raw().get(key));
   }
 
+  default <T> List<T> values(String key, Class<T> clz) {
+    return new Gson()
+        .fromJson(value(key), TypeToken.getParameterized(ArrayList.class, clz).getType());
+  }
+
+  default <T> T value(String key, Class<T> clz) {
+    return new Gson().fromJson(value(key), TypeToken.get(clz).getType());
+  }
+
   default String value(String key) {
     var value = raw().get(key);
     if (value == null) throw new NoSuchElementException("the value for " + key + " is nonexistent");
diff --git a/app/src/main/java/org/astraea/app/web/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
index c3e6128d..2c8fb232 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -18,6 +18,7 @@ package org.astraea.app.web;
 
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toList;
+import static org.astraea.app.web.PostRequest.handleDouble;
 
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonDeserializationContext;
@@ -36,6 +37,9 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.argument.DurationField;
@@ -44,15 +48,12 @@ import org.astraea.app.consumer.Builder;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
 import org.astraea.app.producer.Producer;
+import org.astraea.app.producer.Sender;
 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 RECORDS = "records";
+  static final String TRANSACTION_ID = "transactionId";
   static final String PARTITION = "partition";
   static final String ASYNC = "async";
   static final String DISTANCE_FROM_LATEST = "distanceFromLatest";
@@ -60,7 +61,7 @@ public class RecordHandler implements Handler {
   static final String SEEK_TO = "seekTo";
   static final String KEY_DESERIALIZER = "keyDeserializer";
   static final String VALUE_DESERIALIZER = "valueDeserializer";
-  static final String RECORDS = "records";
+  static final String LIMIT = "limit";
   static final String TIMEOUT = "timeout";
 
   final String bootstrapServers;
@@ -128,37 +129,72 @@ public class RecordHandler implements Handler {
         .ifPresent(seekTo -> consumerBuilder.seekStrategy(Builder.SeekStrategy.SEEK_TO, seekTo));
 
     try (var consumer = consumerBuilder.build()) {
-      var maxRecords = Integer.parseInt(queries.getOrDefault(RECORDS, "1"));
+      var limit = Integer.parseInt(queries.getOrDefault(LIMIT, "1"));
       return new Records(
-          consumer.poll(maxRecords, timeout).stream()
+          consumer.poll(limit, timeout).stream()
               .map(Record::new)
               // TODO: remove limit here (https://github.com/skiptests/astraea/issues/441)
-              .limit(maxRecords)
+              .limit(limit)
               .collect(toList()));
     }
   }
 
   @Override
   public Response post(PostRequest request) {
-    var topic =
-        request.get(TOPIC).orElseThrow(() -> new IllegalArgumentException("topic must be set"));
-    var keySerializer = request.get(KEY_SERIALIZER).map(SerDe::of).orElse(SerDe.STRING).serializer;
-    var valueSerializer =
-        request.get(VALUE_SERIALIZER).map(SerDe::of).orElse(SerDe.STRING).serializer;
     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(keySerializer.apply(topic, k)));
-    request.get(VALUE).ifPresent(v -> sender.value(valueSerializer.apply(topic, v)));
-    request.get(TIMESTAMP).ifPresent(t -> sender.timestamp(Long.parseLong(t)));
-    request.get(PARTITION).ifPresent(p -> sender.partition(Integer.parseInt(p)));
+    var timeout = request.get(TIMEOUT).map(DurationField::toDuration).orElse(Duration.ofSeconds(5));
+    var records = request.values(RECORDS, PostRecord.class);
+    if (records.isEmpty()) {
+      throw new IllegalArgumentException("records should contain at least one record");
+    }
 
-    var senderFuture = sender.run().toCompletableFuture();
+    var producer =
+        request
+            .get(TRANSACTION_ID)
+            .map(
+                // TODO: Find a way to cache transactional producer
+                // (https://github.com/skiptests/astraea/issues/473)
+                transactionId ->
+                    Producer.builder()
+                        .transactionId(transactionId)
+                        .bootstrapServers(bootstrapServers)
+                        .buildTransactional())
+            .orElse(this.producer);
+
+    var result =
+        CompletableFuture.supplyAsync(
+                () -> {
+                  try {
+                    return producer.send(
+                        records.stream()
+                            .map(record -> createSender(producer, record))
+                            .collect(toList()));
+                  } finally {
+                    if (producer.transactional()) {
+                      producer.close();
+                    }
+                  }
+                })
+            .thenCompose(
+                senderFutures ->
+                    Utils.sequence(
+                        senderFutures.stream()
+                            .map(
+                                s ->
+                                    s.handle(
+                                        (metadata, exception) -> {
+                                          if (metadata != null) return new Metadata(metadata);
+                                          if (exception != null)
+                                            return Response.for500(exception.getMessage());
+                                          // this shouldn't happen, but still add error 404 here
+                                          return Response.for404("missing result");
+                                        }))
+                            .map(CompletionStage::toCompletableFuture)
+                            .collect(toList())));
 
     if (async) return Response.ACCEPT;
-
-    return Utils.packException(() -> new Metadata(senderFuture.get()));
+    return Utils.packException(
+        () -> new PostResponse(result.get(timeout.toNanos(), TimeUnit.NANOSECONDS)));
   }
 
   enum SerDe {
@@ -214,6 +250,32 @@ public class RecordHandler implements Handler {
     }
   }
 
+  private static Sender<byte[], byte[]> createSender(
+      Producer<byte[], byte[]> producer, PostRecord postRecord) {
+    var topic =
+        Optional.ofNullable(postRecord.topic)
+            .orElseThrow(() -> new IllegalArgumentException("topic must be set"));
+    var sender = producer.sender().topic(topic);
+    // TODO: Support headers
+    // (https://github.com/skiptests/astraea/issues/422)
+    var keySerializer =
+        Optional.ofNullable(postRecord.keySerializer)
+            .map(name -> SerDe.of(name).serializer)
+            .orElse(SerDe.STRING.serializer);
+    var valueSerializer =
+        Optional.ofNullable(postRecord.valueSerializer)
+            .map(name -> SerDe.of(name).serializer)
+            .orElse(SerDe.STRING.serializer);
+
+    Optional.ofNullable(postRecord.key)
+        .ifPresent(key -> sender.key(keySerializer.apply(topic, handleDouble(key))));
+    Optional.ofNullable(postRecord.value)
+        .ifPresent(value -> sender.value(valueSerializer.apply(topic, handleDouble(value))));
+    Optional.ofNullable(postRecord.timestamp).ifPresent(sender::timestamp);
+    Optional.ofNullable(postRecord.partition).ifPresent(sender::partition);
+    return sender;
+  }
+
   static class Metadata implements Response {
     final String topic;
     final int partition;
@@ -297,4 +359,39 @@ public class RecordHandler implements Handler {
       return new JsonPrimitive(Base64.getEncoder().encodeToString(src));
     }
   }
+
+  static class PostRecord {
+    final String topic;
+    final Integer partition;
+    final String keySerializer;
+    final String valueSerializer;
+    final Object key;
+    final Object value;
+    final Long timestamp;
+
+    PostRecord(
+        String topic,
+        Integer partition,
+        String keySerializer,
+        String valueSerializer,
+        Object key,
+        Object value,
+        Long timestamp) {
+      this.topic = topic;
+      this.partition = partition;
+      this.keySerializer = keySerializer;
+      this.valueSerializer = valueSerializer;
+      this.key = key;
+      this.value = value;
+      this.timestamp = timestamp;
+    }
+  }
+
+  static class PostResponse implements Response {
+    final List<Response> results;
+
+    PostResponse(List<Response> results) {
+      this.results = results;
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/web/Response.java b/app/src/main/java/org/astraea/app/web/Response.java
index a7d54d74..2184ded8 100644
--- a/app/src/main/java/org/astraea/app/web/Response.java
+++ b/app/src/main/java/org/astraea/app/web/Response.java
@@ -34,6 +34,10 @@ interface Response {
     return new ResponseImpl(404, message);
   }
 
+  static Response for500(String message) {
+    return new ResponseImpl(500, message);
+  }
+
   private static int code(Exception exception) {
     if (exception instanceof IllegalArgumentException) return 400;
     if (exception instanceof NoSuchElementException) return 404;
diff --git a/docs/web_server/web_api_records_chinese.md b/docs/web_server/web_api_records_chinese.md
index a6281e91..a477f13d 100644
--- a/docs/web_server/web_api_records_chinese.md
+++ b/docs/web_server/web_api_records_chinese.md
@@ -14,20 +14,29 @@ POST /records
 
 參數
 
+| 名稱            | 說明                                                                                                           | 預設值   |
+|---------------|--------------------------------------------------------------------------------------------------------------|-------|
+| records       | (必填) 本次寫入資料，需填入至少一筆資料                                                                                        | 無     |
+| transactionId | (選填) 填入 transaction id 以執行 transaction 寫入，此值為字串                                                              | 無     |
+| async         | (選填) 是否要等到資料寫入 topic 再回傳                                                                                     | false |
+| timeout       | (選填) 寫入資料最大時限，逾時便中斷本次操作。格式為 `數值` + `單位`，單位可填寫 `days`, `day`, `h`, `m`, `s`, `ms`, `us`, `ns`。範例：1h，5s，1000ms | 5s    |
+
+- 若 async = true，且操作成功，僅回傳 HTTP 202
+
+records 每筆資料欄位
+
 | 名稱              | 說明                        | 預設值    |
 |-----------------|---------------------------|--------|
 | topic           | (必填) topic 名稱             | 無      |
 | partition       | (選填) 指定要寫入此筆資料的 partition | 無      |
 | keySerializer   | (選填) key serializer       | string |
 | valueSerializer | (選填) value serializer     | string |
-| async           | (選填) 是否要等到資料寫入 topic 再回傳  | false  |
 | key             | (選填) key 值                | null   |
 | value           | (選填) value 值              | null   |
 | timestamp       | (選填) 此筆資料時間戳記             | null   |
 
 - keySerializer/valueSerializer 可選擇 `bytearray`/`strinng`/`long`/`integer`/`float`/`double`
 - 若 serializer 選擇 `bytearray`，需將值為 byte array 的 key/value 做 base64 encoding 轉字串
-- 若 async = true，且操作成功，僅回傳 HTTP 202
 
 cURL 範例
 
@@ -35,15 +44,28 @@ cURL 範例
 curl -X POST http://localhost:8001/records \
     -H "Content-Type: application/json" \
     -d '{
-    "topic": "test1",
-    "partition": 0,
-    "keySerializer": "string",
-    "valueSerializer": "bytearray",
     "async": false,
-    "key": "test",
-    "value": "dGVzdA==", #此為 "test" 經 base64 encoding 後之字串
-    "timestamp": 1656337829
-    }'
+    "transactionId": "trx-1",
+    "records": [
+      {
+        "topic": "test1",
+        "partition": 0,
+        "keySerializer": "string",
+        "valueSerializer": "bytearray",
+        "key": "test",
+        "value": "dGVzdA==", #此為 "test" 經 base64 encoding 後之字串
+        "timestamp": 1656337829
+      },
+      {
+        "topic": "test1",
+        "partition": 0,
+        "keySerializer": "string",
+        "valueSerializer": "bytearray",
+        "key": "test",
+        "value": "dGVzdA==", #此為 "test" 經 base64 encoding 後之字串
+        "timestamp": 1656337829
+      }
+    ]}'
 ```
 JSON Response (僅在 async = false 時回傳)
 
@@ -56,12 +78,24 @@ JSON Response (僅在 async = false 時回傳)
 
 ```json
 {
-  "topic": "test1",
-  "partition": 0,
-  "offset": 0,
-  "timestamp": 1656337829,
-  "serializedKeySize": 4,
-  "serializedValueSize": 4
+  "results": [
+    {
+      "topic": "test1",
+      "partition": 0,
+      "offset": 0,
+      "timestamp": 1656337829,
+      "serializedKeySize": 4,
+      "serializedValueSize": 4
+    },
+    {
+      "topic": "test1",
+      "partition": 0,
+      "offset": 1,
+      "timestamp": 1656337829,
+      "serializedKeySize": 4,
+      "serializedValueSize": 4
+    }
+  ]
 }
 ```
 
@@ -78,7 +112,7 @@ GET /records/{topic}
 | partition             | (選填) 指定要讀取之 partition                                                                                        | 無      |
 | keyDeserializer       | (選填) key deserializer                                                                                        | string |
 | valueDeserializer     | (選填) value deserializer                                                                                      | string |
-| records               | (選填) 回傳資料筆數上限                                                                                                | 1      |
+| limit                 | (選填) 回傳資料筆數上限                                                                                                | 1      |
 | timeout               | (選填) 請求資料最大時限，逾時便中斷本次操作。格式為 `數值` + `單位`，單位可填寫 `days`, `day`, `h`, `m`, `s`, `ms`, `us`, `ns`。範例：1h，5s，1000ms | 5s     |
 | distanceFromLatest    | (選填) 距離最新 offset 往前多少位移量開始拉取資料                                                                               | 無      |
 | distanceFromBeginning | (選填) 距離起始 offset 往後多少位移量開始拉取資料                                                                               | 無      |
@@ -98,7 +132,7 @@ GET /records/{topic}
 cURL 範例
 
 ```shell
-curl -X GET "http://localhost:8001/records/test?distanceFromLatest=3&records=2&keyDeserializer=string&valueDeserializer=integer"
+curl -X GET "http://localhost:8001/records/test?distanceFromLatest=3&limit=2&keyDeserializer=string&valueDeserializer=integer"
 ```
 
 JSON Response
@@ -152,7 +186,7 @@ JSON Response
 cURL 範例
 
 ```shell
-curl -X GET "http://localhost:8001/records/test?distanceFromBeginning=3&records=2&keyDeserializer=string&valueDeserializer=integer"
+curl -X GET "http://localhost:8001/records/test?distanceFromBeginning=3&limit=2&keyDeserializer=string&valueDeserializer=integer"
 ```
 
 JSON Response
@@ -195,7 +229,7 @@ JSON Response
 cURL 範例 - 從 offset=1 開始拉取共兩筆資料
 
 ```shell
-curl -X GET "http://localhost:8001/records/test?seekTo=1&records=2&keyDeserializer=string&valueDeserializer=integer"
+curl -X GET "http://localhost:8001/records/test?seekTo=1&limit=2&keyDeserializer=string&valueDeserializer=integer"
 ```
 
 JSON Response


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 fd8f2d9d..9de9b0d6 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -27,7 +27,6 @@ import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -120,12 +119,12 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testPartitions() throws InterruptedException {
+  void testPartitions() {
     var topicName = "testPartitions";
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(3).create();
       // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       Assertions.assertTrue(admin.topicNames().contains(topicName));
       var partitions = admin.replicas(Set.of(topicName));
       Assertions.assertEquals(3, partitions.size());
@@ -143,12 +142,12 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testOffsets() throws InterruptedException {
+  void testOffsets() {
     var topicName = "testOffsets";
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(3).create();
       // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       var offsets = admin.offsets(Set.of(topicName));
       Assertions.assertEquals(3, offsets.size());
       offsets
@@ -162,7 +161,7 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testConsumerGroups() throws InterruptedException {
+  void testConsumerGroups() {
     var topicName = "testConsumerGroups-Topic";
     var consumerGroup = "testConsumerGroups-Group";
     try (var admin = Admin.of(bootstrapServers())) {
@@ -173,7 +172,7 @@ public class AdminTest extends RequireBrokerCluster {
               .groupId(consumerGroup)
               .build()) {
         // wait for syncing topic creation
-        TimeUnit.SECONDS.sleep(5);
+        Utils.sleep(Duration.ofSeconds(5));
         var consumerGroupMap = admin.consumerGroups(Set.of(consumerGroup));
         Assertions.assertEquals(1, consumerGroupMap.size());
         Assertions.assertTrue(consumerGroupMap.containsKey(consumerGroup));
@@ -199,12 +198,12 @@ public class AdminTest extends RequireBrokerCluster {
   // There is a problem when migrating the log folder under Windows because the migrated source
   // cannot be deleted, so disabled this test on Windows for now.
   @DisabledOnOs(WINDOWS)
-  void testMigrateSinglePartition() throws InterruptedException {
+  void testMigrateSinglePartition() {
     var topicName = "testMigrateSinglePartition";
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(1).create();
       // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       var broker = admin.brokerIds().iterator().next();
       admin.migrator().partition(topicName, 0).moveTo(List.of(broker));
       Utils.waitFor(
@@ -246,13 +245,13 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testDeclarePreferredLogDirectory() throws InterruptedException {
+  void testDeclarePreferredLogDirectory() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = "DeclarePreferredLogDirectory_" + Utils.randomString();
       var topicPartition = new TopicPartition(topic, 0);
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 1).create();
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
       var originalBroker = admin.replicas(Set.of(topic)).get(topicPartition).get(0).broker();
       var nextBroker = (originalBroker + 1) % brokerIds().size();
       var nextDir = logFolders().get(nextBroker).stream().findAny().orElseThrow();
@@ -263,7 +262,7 @@ public class AdminTest extends RequireBrokerCluster {
           () ->
               admin.migrator().partition(topic, 0).declarePreferredDir(Map.of(nextBroker, nextDir)),
           "The Migrator API should ignore the error");
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
 
       // assert, nothing happened until the actual movement
       Assertions.assertNotEquals(nextBroker, replicaNow.get().broker());
@@ -271,7 +270,7 @@ public class AdminTest extends RequireBrokerCluster {
 
       // act, perform the actual movement
       admin.migrator().partition(topic, 0).moveTo(List.of(nextBroker));
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
 
       // assert, everything on the exact broker & dir
       Assertions.assertEquals(nextBroker, replicaNow.get().broker());
@@ -280,12 +279,12 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testIllegalMigrationArgument() throws InterruptedException {
+  void testIllegalMigrationArgument() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = Utils.randomString();
       var topicPartition = new TopicPartition(topic, 0);
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 1).create();
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
       var currentReplica = admin.replicas(Set.of(topic)).get(topicPartition).get(0);
       var currentBroker = currentReplica.broker();
       var notExistReplica = (currentBroker + 1) % brokerIds().size();
@@ -298,12 +297,12 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testIllegalPreferredDirArgument() throws InterruptedException {
+  void testIllegalPreferredDirArgument() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = Utils.randomString();
       var topicPartition = new TopicPartition(topic, 0);
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 1).create();
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
       var currentReplica = admin.replicas(Set.of(topic)).get(topicPartition).get(0);
       var currentBroker = currentReplica.broker();
       var nextDir = logFolders().get(currentBroker).iterator().next();
@@ -319,12 +318,12 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testPartialMoveToArgument() throws InterruptedException {
+  void testPartialMoveToArgument() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = Utils.randomString();
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 3).create();
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
       var replica0 = 0;
       var replica1 = 1;
       var replica2 = 2;
@@ -352,12 +351,12 @@ public class AdminTest extends RequireBrokerCluster {
 
   @Test
   @DisabledOnOs(WINDOWS)
-  void testMigrateAllPartitions() throws InterruptedException {
+  void testMigrateAllPartitions() {
     var topicName = "testMigrateAllPartitions";
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(3).create();
       // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       var broker = admin.brokerIds().iterator().next();
       admin.migrator().topic(topicName).moveTo(List.of(broker));
       Utils.waitFor(
@@ -390,7 +389,7 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testCompact() throws InterruptedException {
+  void testCompact() {
     var topicName = "testCompacted";
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).compactionMaxLag(Duration.ofSeconds(1)).create();
@@ -409,7 +408,7 @@ public class AdminTest extends RequireBrokerCluster {
         producer.flush();
 
         // sleep and produce more data to generate the new segment
-        TimeUnit.SECONDS.sleep(2);
+        Utils.sleep(Duration.ofSeconds(2));
         IntStream.range(0, 10)
             .forEach(i -> producer.sender().key(anotherKey).value(value).topic(topicName).run());
         producer.flush();
@@ -417,7 +416,7 @@ public class AdminTest extends RequireBrokerCluster {
 
       // sleep for compact (the backoff of compact thread is reduced to 2 seconds.
       // see org.astraea.service.Services)
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       try (var consumer =
           Consumer.forTopics(Set.of(topicName))
@@ -473,10 +472,10 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testReplicas() throws InterruptedException {
+  void testReplicas() {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic("abc").numberOfPartitions(2).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       Assertions.assertEquals(2, admin.replicas(Set.of("abc")).size());
 
       var count = admin.topicNames().stream().mapToInt(t -> admin.replicas(Set.of(t)).size()).sum();
@@ -485,7 +484,7 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testReplicasPreferredLeaderFlag() throws InterruptedException {
+  void testReplicasPreferredLeaderFlag() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = "testReplicasPreferredLeaderFlag_" + Utils.randomString();
@@ -496,7 +495,7 @@ public class AdminTest extends RequireBrokerCluster {
           .numberOfPartitions(partitionCount)
           .numberOfReplicas((short) 3)
           .create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       var expectedPreferredLeader =
           IntStream.range(0, partitionCount)
               .mapToObj(p -> new TopicPartition(topic, p))
@@ -517,7 +516,7 @@ public class AdminTest extends RequireBrokerCluster {
       // act, make 0 be the preferred leader of every partition
       IntStream.range(0, partitionCount)
           .forEach(p -> admin.migrator().partition(topic, p).moveTo(List.of(0, 1, 2)));
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       // assert
       Assertions.assertEquals(expectedPreferredLeader, currentPreferredLeader.get());
@@ -543,10 +542,10 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testIpQuota() throws InterruptedException {
+  void testIpQuota() {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.quotaCreator().ip("192.168.11.11").connectionRate(10).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       java.util.function.Consumer<List<Quota>> checker =
           (quotas) -> {
@@ -573,21 +572,21 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testMultipleIpQuota() throws InterruptedException {
+  void testMultipleIpQuota() {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.quotaCreator().ip("192.168.11.11").connectionRate(10).create();
       admin.quotaCreator().ip("192.168.11.11").connectionRate(12).create();
       admin.quotaCreator().ip("192.168.11.11").connectionRate(9).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       Assertions.assertEquals(1, admin.quotas(Quota.Target.IP, "192.168.11.11").size());
     }
   }
 
   @Test
-  void testClientQuota() throws InterruptedException {
+  void testClientQuota() {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.quotaCreator().clientId("my-id").produceRate(10).consumeRate(100).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       java.util.function.Consumer<List<Quota>> checker =
           (quotas) -> {
@@ -640,11 +639,11 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testMultipleClientQuota() throws InterruptedException {
+  void testMultipleClientQuota() {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.quotaCreator().clientId("my-id").consumeRate(100).create();
       admin.quotaCreator().clientId("my-id").produceRate(999).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       Assertions.assertEquals(2, admin.quotas(Quota.Target.CLIENT_ID, "my-id").size());
     }
   }
@@ -659,7 +658,7 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testClusterInfo() throws InterruptedException {
+  void testClusterInfo() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       String topic0 = "testClusterInfoFromAdmin_" + Utils.randomString(8);
       String topic1 = "testClusterInfoFromAdmin_" + Utils.randomString(8);
@@ -676,7 +675,7 @@ public class AdminTest extends RequireBrokerCluster {
                       .numberOfPartitions(partitionCount)
                       .numberOfReplicas(replicaCount)
                       .create());
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       final var clusterInfo = admin.clusterInfo(Set.of(topic0, topic1, topic2));
 
@@ -721,12 +720,12 @@ public class AdminTest extends RequireBrokerCluster {
 
   @ParameterizedTest
   @ValueSource(shorts = {1, 2, 3})
-  void preferredLeaderElection(short replicaSize) throws InterruptedException {
+  void preferredLeaderElection(short replicaSize) {
     var clusterSize = brokerIds().size();
     var topic = "preferredLeaderElection_" + Utils.randomString(6);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(30).numberOfReplicas(replicaSize).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       var topicPartitions =
           IntStream.range(0, 30)
@@ -775,7 +774,7 @@ public class AdminTest extends RequireBrokerCluster {
                   .migrator()
                   .partition(topicPartition.topic(), topicPartition.partition())
                   .moveTo(expectedReplicaList.get(topicPartition)));
-      TimeUnit.SECONDS.sleep(8);
+      Utils.sleep(Duration.ofSeconds(8));
 
       // ReplicaMigrator#moveTo will trigger leader election if current leader being kicked out of
       // replica list. This case is always true for replica size equals to 1.
@@ -787,7 +786,7 @@ public class AdminTest extends RequireBrokerCluster {
 
         // act, the Admin#preferredLeaderElection won't throw a ElectionNotNeededException
         topicPartitions.forEach(admin::preferredLeaderElection);
-        TimeUnit.SECONDS.sleep(2);
+        Utils.sleep(Duration.ofSeconds(2));
 
         // after election
         Assertions.assertEquals(expectedLeaderMap.get(), currentLeaderMap.get());
@@ -797,7 +796,7 @@ public class AdminTest extends RequireBrokerCluster {
 
         // act
         topicPartitions.forEach(admin::preferredLeaderElection);
-        TimeUnit.SECONDS.sleep(2);
+        Utils.sleep(Duration.ofSeconds(2));
 
         // after election
         Assertions.assertEquals(expectedLeaderMap.get(), currentLeaderMap.get());
@@ -950,11 +949,11 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @RepeatedTest(5) // this test is timer-based, so it needs to be looped for stability
-  void testReassignmentWhenMovingPartitionToAnotherBroker() throws InterruptedException {
+  void testReassignmentWhenMovingPartitionToAnotherBroker() {
     var topicName = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(1).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       var currentBroker =
           admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0).broker();
@@ -995,11 +994,11 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @RepeatedTest(5) // this test is timer-based, so it needs to be looped for stability
-  void testReassignmentWhenMovingPartitionToAnotherPath() throws InterruptedException {
+  void testReassignmentWhenMovingPartitionToAnotherPath() {
     var topicName = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(1).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       var currentReplica =
           admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0);
@@ -1047,11 +1046,11 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @RepeatedTest(5) // this test is timer-based, so it needs to be looped for stability
-  void testMultiReassignments() throws InterruptedException {
+  void testMultiReassignments() {
     var topicName = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(1).numberOfReplicas((short) 3).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       var brokers = new ArrayList<>(brokerIds()).subList(0, 2);
       try (var producer = Producer.of(bootstrapServers())) {
@@ -1084,11 +1083,11 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testReassignmentWithNothing() throws InterruptedException {
+  void testReassignmentWithNothing() {
     var topicName = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(1).numberOfReplicas((short) 3).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       var brokers = new ArrayList<>(brokerIds()).subList(0, 2);
       try (var producer = Producer.of(bootstrapServers())) {
         producer.sender().topic(topicName).value(new byte[100]).run();
diff --git a/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
index ae6f1b45..41141390 100644
--- a/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
@@ -16,9 +16,9 @@
  */
 package org.astraea.app.admin;
 
+import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -26,7 +26,7 @@ import org.junit.jupiter.api.Test;
 
 public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
   @Test
-  void testClusterInfoWithOfflineNode() throws InterruptedException {
+  void testClusterInfoWithOfflineNode() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topicName = "ClusterInfo_Offline_" + Utils.randomString();
       var partitionCount = 30;
@@ -37,7 +37,7 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
           .numberOfPartitions(partitionCount)
           .numberOfReplicas(replicaCount)
           .create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       // before node offline
       var before = admin.clusterInfo(Set.of(topicName));
@@ -51,7 +51,7 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
       // act
       int brokerToClose = ThreadLocalRandom.current().nextInt(0, 3);
       closeBroker(brokerToClose);
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
 
       // after node offline
       var after = admin.clusterInfo(Set.of(topicName));
diff --git a/app/src/test/java/org/astraea/app/admin/ReplicaCollieTest.java b/app/src/test/java/org/astraea/app/admin/ReplicaCollieTest.java
index 053f9ee4..5fad277a 100644
--- a/app/src/test/java/org/astraea/app/admin/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ReplicaCollieTest.java
@@ -18,11 +18,11 @@ package org.astraea.app.admin;
 
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.astraea.app.argument.Argument;
 import org.astraea.app.common.Utils;
@@ -36,19 +36,19 @@ import org.junit.jupiter.api.condition.DisabledOnOs;
 public class ReplicaCollieTest extends RequireBrokerCluster {
   @Test
   @DisabledOnOs(WINDOWS)
-  void testVerify() throws InterruptedException {
+  void testVerify() {
     test(true);
   }
 
   @Test
   @DisabledOnOs(WINDOWS)
-  void testExecute() throws InterruptedException {
+  void testExecute() {
     test(false);
   }
 
   @Test
   @DisabledOnOs(WINDOWS)
-  void testBrokerMigrator() throws InterruptedException {
+  void testBrokerMigrator() {
     var topicName = "ReplicaCollieTest-Broker";
     try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
@@ -58,7 +58,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
           .numberOfReplicas((short) 2)
           .create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       var partitionReplicas = topicAdmin.replicas(Set.of(topicName));
       Assertions.assertEquals(1, partitionReplicas.size());
       var brokerSource =
@@ -88,7 +88,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
 
   @Test
   @DisabledOnOs(WINDOWS)
-  void testPathMigrator() throws InterruptedException {
+  void testPathMigrator() {
     var topicName = "ReplicaCollieTest-Path";
     try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
@@ -98,7 +98,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
           .numberOfReplicas((short) 2)
           .create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       var partitionReplicas = topicAdmin.replicas(Set.of(topicName));
       Assertions.assertEquals(1, partitionReplicas.size());
       var brokerSource = partitionReplicas.get(new TopicPartition(topicName, 0)).get(0).broker();
@@ -129,7 +129,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
     }
   }
 
-  private void test(boolean verify) throws InterruptedException {
+  private void test(boolean verify) {
     var topicName = "ReplicaCollieTest-" + verify;
     try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
@@ -139,7 +139,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
           .numberOfReplicas((short) 1)
           .create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       var partitionReplicas = topicAdmin.replicas(Set.of(topicName));
       Assertions.assertEquals(1, partitionReplicas.size());
       var replicas =
diff --git a/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java b/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java
index 8c23b1f2..a594a26d 100644
--- a/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java
@@ -20,12 +20,14 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
 import org.astraea.app.argument.Argument;
+import org.astraea.app.common.Utils;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.producer.Sender;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -85,7 +87,7 @@ class ReplicaSyncingMonitorIntegrationTest extends RequireBrokerCluster {
       // act
       executionThread.start();
       TimeUnit.SECONDS.timedJoin(executionThread, 8); // wait until the thread exit
-      TimeUnit.SECONDS.sleep(2); // sleep 2 extra seconds to ensure test run in stable
+      Utils.sleep(Duration.ofSeconds(2)); // sleep 2 extra seconds to ensure test run in stable
 
       // assert
       assertSame(Thread.State.TERMINATED, executionThread.getState());
diff --git a/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java b/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
index bec53857..2b68166c 100644
--- a/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
+++ b/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
@@ -16,9 +16,10 @@
  */
 package org.astraea.app.admin;
 
+import java.time.Duration;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -32,7 +33,7 @@ public class SomePartitionOfflineTest extends RequireBrokerCluster {
       admin.creator().topic(topicName1).numberOfPartitions(4).numberOfReplicas((short) 1).create();
       admin.creator().topic(topicName2).numberOfPartitions(4).numberOfReplicas((short) 1).create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(10);
+      Utils.sleep(Duration.ofSeconds(3));
       var replicaOnBroker0 =
           admin.replicas(admin.topicNames()).entrySet().stream()
               .filter(replica -> replica.getValue().get(0).broker() == 0)
@@ -48,9 +49,6 @@ public class SomePartitionOfflineTest extends RequireBrokerCluster {
               .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
       offlineReplicaOnBroker0.forEach(
           (tp, replica) -> Assertions.assertTrue(replica.get(0).isOffline()));
-
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java b/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java
index eb5255d7..9384e221 100644
--- a/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java
+++ b/app/src/test/java/org/astraea/app/admin/TopicExplorerTest.java
@@ -32,9 +32,9 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterEach;
@@ -96,7 +96,7 @@ public class TopicExplorerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testExecute() throws InterruptedException {
+  void testExecute() {
     // arrange
     var topicName = "TopicExplorerTest-testExecute-Topic";
     var groupName = "TopicExplorerTest-testExecute-ConsumerGroup";
@@ -106,7 +106,7 @@ public class TopicExplorerTest extends RequireBrokerCluster {
       consumer(Set.of(topicName), groupName, "beta");
       consumer(Set.of(topicName), groupName, "gamma");
 
-      TimeUnit.SECONDS.sleep(10); // wait for broker process resource creation
+      Utils.sleep(Duration.ofSeconds(5)); // wait for broker process resource creation
 
       // act
       var result = TopicExplorer.execute(admin, Set.of(topicName));
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
index 95ed015d..0d33a8c7 100644
--- a/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
@@ -48,7 +48,7 @@ import org.junit.jupiter.api.Test;
 class RebalanceAdminTest extends RequireBrokerCluster {
 
   @Test
-  void alterReplicaPlacementByList() throws InterruptedException {
+  void alterReplicaPlacementByList() {
     // arrange
     try (var admin = Admin.of(bootstrapServers())) {
       var topic = prepareTopic(admin, 1, (short) 1);
@@ -85,7 +85,7 @@ class RebalanceAdminTest extends RequireBrokerCluster {
 
   // repeat the test so it has higher chance to fail
   @RepeatedTest(value = 3)
-  void alterReplicaPlacementByDirectory() throws InterruptedException {
+  void alterReplicaPlacementByDirectory() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = prepareTopic(admin, 1, (short) 1);
@@ -117,7 +117,7 @@ class RebalanceAdminTest extends RequireBrokerCluster {
   }
 
   @Test
-  void clusterInfo() throws InterruptedException {
+  void clusterInfo() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       // test if all topics are covered
       final var rebalanceAdmin = RebalanceAdmin.of(admin, (ignore) -> true);
@@ -286,7 +286,7 @@ class RebalanceAdminTest extends RequireBrokerCluster {
     }
   }
 
-  String prepareTopic(Admin admin, int partition, short replica) throws InterruptedException {
+  String prepareTopic(Admin admin, int partition, short replica) {
     var topicName = Utils.randomString();
 
     admin
@@ -295,7 +295,7 @@ class RebalanceAdminTest extends RequireBrokerCluster {
         .numberOfPartitions(partition)
         .numberOfReplicas(replica)
         .create();
-    TimeUnit.SECONDS.sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
 
     return topicName;
   }
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
index 59873bfa..fbb7280c 100644
--- a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
@@ -16,9 +16,9 @@
  */
 package org.astraea.app.balancer.executor;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
@@ -34,15 +34,15 @@ import org.junit.jupiter.api.Test;
 class StraightPlanExecutorTest extends RequireBrokerCluster {
 
   @Test
-  void testRun() throws InterruptedException {
+  void testRun() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
       admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 2).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       final var originalAllocation =
           LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       final var broker0 = 0;
       final var broker1 = 1;
       final var logFolder0 = logFolders().get(broker0).stream().findAny().orElseThrow();
@@ -84,15 +84,15 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testRunNoDirSpecified() throws InterruptedException {
+  void testRunNoDirSpecified() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
       admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 3).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       final var originalAllocation =
           LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       final var broker0 = 0;
       final var broker1 = 1;
       final var broker2 = 2;
diff --git a/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java b/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
index 575edab8..16a28918 100644
--- a/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
@@ -16,11 +16,12 @@
  */
 package org.astraea.app.balancer.generator;
 
+import java.time.Duration;
 import java.util.concurrent.ForkJoinPool;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Collectors;
 import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.app.common.Utils;
 import org.astraea.app.cost.ClusterInfoProvider;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Disabled;
@@ -171,14 +172,9 @@ class ShufflePlanGeneratorTest {
         new Thread(
             () -> {
               while (!Thread.currentThread().isInterrupted()) {
-                try {
-                  TimeUnit.SECONDS.sleep(1);
-                  long now = System.nanoTime();
-                  System.out.printf(
-                      "%.3f proposal/s %n", counter.sum() / ((now - startTime) / 1e9));
-                } catch (InterruptedException e) {
-                  break;
-                }
+                Utils.sleep(Duration.ofSeconds(1));
+                long now = System.nanoTime();
+                System.out.printf("%.3f proposal/s %n", counter.sum() / ((now - startTime) / 1e9));
               }
             });
 
diff --git a/app/src/test/java/org/astraea/app/common/UtilsTest.java b/app/src/test/java/org/astraea/app/common/UtilsTest.java
index fdbdc6bc..8b907948 100644
--- a/app/src/test/java/org/astraea/app/common/UtilsTest.java
+++ b/app/src/test/java/org/astraea/app/common/UtilsTest.java
@@ -17,8 +17,10 @@
 package org.astraea.app.common;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.SortedMap;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -66,4 +68,12 @@ public class UtilsTest {
             Stream.of(Map.entry(1, "hello"), Map.entry(1, "world"))
                 .collect(Utils.toSortedMap(Map.Entry::getKey, Map.Entry::getValue)));
   }
+
+  @Test
+  void testSequence() {
+    var future1 = CompletableFuture.supplyAsync(() -> 1);
+    var future2 = CompletableFuture.supplyAsync(() -> 2);
+
+    Assertions.assertEquals(Utils.sequence(List.of(future1, future2)).join(), List.of(1, 2));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/concurrent/ThreadPoolTest.java b/app/src/test/java/org/astraea/app/concurrent/ThreadPoolTest.java
index 9ab6c645..18f2e1af 100644
--- a/app/src/test/java/org/astraea/app/concurrent/ThreadPoolTest.java
+++ b/app/src/test/java/org/astraea/app/concurrent/ThreadPoolTest.java
@@ -16,8 +16,9 @@
  */
 package org.astraea.app.concurrent;
 
-import java.util.concurrent.TimeUnit;
+import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.astraea.app.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
@@ -51,7 +52,7 @@ public class ThreadPoolTest {
   void testSubmitThread() throws Exception {
     var executor = new CountExecutor();
     try (var pool = ThreadPool.builder().executor(executor).build()) {
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
     }
     Assertions.assertTrue(executor.executeCount.get() > 0);
     Assertions.assertEquals(1, executor.closeCount.get());
@@ -72,7 +73,7 @@ public class ThreadPoolTest {
         ThreadPool.builder()
             .executor(
                 () -> {
-                  TimeUnit.SECONDS.sleep(1000);
+                  Utils.sleep(Duration.ofSeconds(100));
                   return State.DONE;
                 })
             .build();
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 19b72d66..5661b535 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -99,12 +99,8 @@ public class ConsumerTest extends RequireBrokerCluster {
       var service = Executors.newSingleThreadExecutor();
       service.execute(
           () -> {
-            try {
-              TimeUnit.SECONDS.sleep(3);
-              consumer.wakeup();
-            } catch (InterruptedException ignored) {
-              // swallow
-            }
+            Utils.sleep(Duration.ofSeconds(3));
+            consumer.wakeup();
           });
       // this call will be broken after 3 seconds
       Assertions.assertThrows(WakeupException.class, () -> consumer.poll(Duration.ofSeconds(100)));
@@ -208,13 +204,13 @@ public class ConsumerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testAssignment() throws InterruptedException {
+  void testAssignment() {
     var topic = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       var partitionNum = 2;
       admin.creator().topic(topic).numberOfPartitions(partitionNum).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       for (int partitionId = 0; partitionId < partitionNum; partitionId++) {
         for (int recordIdx = 0; recordIdx < 10; recordIdx++) {
@@ -254,12 +250,12 @@ public class ConsumerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testCommitOffset() throws InterruptedException {
+  void testCommitOffset() {
     var topic = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(1).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       producer.sender().topic(topic).value(new byte[10]).run();
       producer.flush();
 
diff --git a/app/src/test/java/org/astraea/app/cost/PeriodicTest.java b/app/src/test/java/org/astraea/app/cost/PeriodicTest.java
index 8aa2cf58..92360e33 100644
--- a/app/src/test/java/org/astraea/app/cost/PeriodicTest.java
+++ b/app/src/test/java/org/astraea/app/cost/PeriodicTest.java
@@ -19,7 +19,7 @@ package org.astraea.app.cost;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
+import org.astraea.app.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -32,7 +32,7 @@ public class PeriodicTest extends Periodic<Map<Integer, Double>> {
     Assertions.assertEquals(broker1.get(0), 0.0);
     var broker2 = tryUpdateAfterOneSecond(this::testMap);
     Assertions.assertEquals(broker2.get(0), 0.0);
-    sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
     broker2 = tryUpdateAfterOneSecond(this::testMap);
     Assertions.assertEquals(broker2.get(0), 1.0);
   }
@@ -43,7 +43,7 @@ public class PeriodicTest extends Periodic<Map<Integer, Double>> {
     Assertions.assertEquals(broker1.get(0), 0.0);
     var broker2 = tryUpdate(this::testMap, Duration.ofSeconds(3));
     Assertions.assertEquals(broker2.get(0), 0.0);
-    sleep(4);
+    Utils.sleep(Duration.ofSeconds(4));
     broker2 = tryUpdate(this::testMap, Duration.ofSeconds(3));
     Assertions.assertEquals(broker2.get(0), 1.0);
   }
@@ -54,12 +54,4 @@ public class PeriodicTest extends Periodic<Map<Integer, Double>> {
     brokerValue++;
     return broker;
   }
-
-  private static void sleep(int seconds) {
-    try {
-      TimeUnit.SECONDS.sleep(seconds);
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-  }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
index f6a9cc2d..368be8f3 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
@@ -23,7 +23,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.ClusterBean;
@@ -31,6 +30,7 @@ import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.broker.HasValue;
@@ -43,14 +43,14 @@ import org.junit.jupiter.api.Test;
 
 class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
   @Test
-  void testGetMetrics() {
+  void testGetMetrics() throws ExecutionException, InterruptedException {
     var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
     var topicName = "testGetMetrics-1";
     try (var admin = Admin.of(bootstrapServers())) {
       var host = "localhost";
       admin.creator().topic(topicName).numberOfPartitions(1).numberOfReplicas((short) 1).create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(5);
+      Utils.sleep(Duration.ofSeconds(5));
       var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
       producer.sender().topic(topicName).key(new byte[10000]).run().toCompletableFuture().get();
       ReplicaSizeCost costFunction = new ReplicaSizeCost(brokerDiskSize);
@@ -75,8 +75,6 @@ class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
               .map(e2 -> (int) e2.value())
               .collect(Collectors.toList());
       Assertions.assertTrue(replicaSize.get(0) >= 10000);
-    } catch (InterruptedException | ExecutionException e) {
-      throw new RuntimeException(e);
     }
   }
 
diff --git a/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java b/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java
index ef9a9ac2..48f1508c 100644
--- a/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java
+++ b/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java
@@ -18,11 +18,12 @@ package org.astraea.app.cost.topic;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
+import org.astraea.app.common.Utils;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.producer.Serializer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -60,9 +61,7 @@ public class GetPartitionInfTest extends RequireBrokerCluster {
           .numberOfReplicas((short) 1)
           .create();
       // wait for topic
-      TimeUnit.SECONDS.sleep(5);
-    } catch (InterruptedException e) {
-      e.printStackTrace();
+      Utils.sleep(Duration.ofSeconds(5));
     }
     var producer =
         Producer.builder()
diff --git a/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java b/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java
index 7cac1304..3298c591 100644
--- a/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java
+++ b/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java
@@ -18,11 +18,12 @@ package org.astraea.app.cost.topic;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
+import org.astraea.app.common.Utils;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.producer.Serializer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -59,9 +60,7 @@ public class PartitionScoreTest extends RequireBrokerCluster {
           .numberOfReplicas((short) 1)
           .create();
       // wait for topic
-      TimeUnit.SECONDS.sleep(5);
-    } catch (InterruptedException e) {
-      e.printStackTrace();
+      Utils.sleep(Duration.ofSeconds(5));
     }
     var producer =
         Producer.builder()
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/KafkaMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/KafkaMetricsTest.java
index 979407d4..e24cfbc6 100644
--- a/app/src/test/java/org/astraea/app/metrics/broker/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/broker/KafkaMetricsTest.java
@@ -25,9 +25,9 @@ import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import javax.management.MBeanServer;
 import javax.management.remote.JMXConnectorServer;
@@ -131,14 +131,14 @@ class KafkaMetricsTest extends RequireBrokerCluster {
 
   @ParameterizedTest()
   @EnumSource(value = KafkaMetrics.TopicPartition.class)
-  void testTopicPartitionMetrics(KafkaMetrics.TopicPartition request) throws InterruptedException {
+  void testTopicPartitionMetrics(KafkaMetrics.TopicPartition request) {
     try (var admin = Admin.of(bootstrapServers())) {
       // there are only 3 brokers, so 10 partitions can make each broker has some partitions
       admin.creator().topic(Utils.randomString(5)).numberOfPartitions(10).create();
     }
 
     // wait for topic creation
-    TimeUnit.SECONDS.sleep(2);
+    Utils.sleep(Duration.ofSeconds(2));
 
     var beans = request.fetch(mBeanClient);
     assertNotEquals(0, beans.size());
diff --git a/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java b/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
index bed62529..ab1bd14c 100644
--- a/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/collector/BeanCollectorTest.java
@@ -20,7 +20,6 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
@@ -42,14 +41,6 @@ public class BeanCollectorTest {
   private final BiFunction<String, Integer, MBeanClient> clientCreator =
       (host, port) -> mbeanClient;
 
-  private static void sleep(int seconds) {
-    try {
-      TimeUnit.SECONDS.sleep(seconds);
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
   private static Executor executor(Runnable runnable) {
     return () -> {
       runnable.run();
@@ -106,12 +97,12 @@ public class BeanCollectorTest {
     var c0 = receiver.current();
     Assertions.assertEquals(1, c0.size());
     var firstObject = c0.iterator().next();
-    sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
 
     var c1 = receiver.current();
     Assertions.assertEquals(2, c1.size());
     var secondObject = c1.stream().filter(o -> o != firstObject).findFirst().get();
-    sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
 
     var c2 = receiver.current();
     Assertions.assertEquals(2, c2.size());
@@ -154,7 +145,7 @@ public class BeanCollectorTest {
               receivers.add(receiver);
             }
           } finally {
-            sleep(1);
+            Utils.sleep(Duration.ofSeconds(1));
           }
         };
 
@@ -165,7 +156,7 @@ public class BeanCollectorTest {
                     .mapToObj(i -> executor(runnable))
                     .collect(Collectors.toList()))
             .build()) {
-      sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
     }
     return receivers;
   }
@@ -204,7 +195,7 @@ public class BeanCollectorTest {
                     .map(receiver -> executor(receiver::current))
                     .collect(Collectors.toList()))
             .build()) {
-      sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
     }
     receivers.forEach(r -> Assertions.assertEquals(1, r.current().size()));
   }
@@ -237,7 +228,7 @@ public class BeanCollectorTest {
     for (var expect : expectedSizes) {
       Assertions.assertEquals(expect, receiver.current().size());
       Assertions.assertEquals(expect, count.get());
-      sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
     }
   }
 
@@ -269,16 +260,16 @@ public class BeanCollectorTest {
                 })
             .collect(Collectors.toList());
 
-    sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
     receivers.forEach(e -> Assertions.assertEquals(0, e.getKey().get()));
 
-    sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
     receivers.forEach(e -> Assertions.assertEquals(1, e.getValue().current().size()));
     receivers.forEach(e -> Assertions.assertEquals(1, e.getKey().get()));
   }
 
   @Test
-  void testCreatedTimestamp() throws InterruptedException {
+  void testCreatedTimestamp() {
     var collector =
         BeanCollector.builder()
             .interval(Duration.ofSeconds(1))
@@ -293,7 +284,7 @@ public class BeanCollectorTest {
             .fetcher(client -> List.of(() -> obj))
             .build()) {
       // wait for updating cache
-      TimeUnit.SECONDS.sleep(1);
+      Utils.sleep(Duration.ofSeconds(1));
       var objs = receiver.current();
       Assertions.assertEquals(1, objs.size());
       Assertions.assertEquals(obj.createdTimestamp(), objs.iterator().next().createdTimestamp());
diff --git a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
index 7dda8f5f..bfc84288 100644
--- a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
@@ -27,7 +27,6 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -35,6 +34,7 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
+import org.astraea.app.common.Utils;
 import org.astraea.app.concurrent.Executor;
 import org.astraea.app.concurrent.State;
 import org.astraea.app.concurrent.ThreadPool;
@@ -118,7 +118,7 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
     } catch (ExecutionException | InterruptedException e) {
       e.printStackTrace();
     }
-    sleep(1);
+    Utils.sleep(Duration.ofSeconds(1));
     try (var consumer =
         Consumer.forTopics(Set.of(topicName))
             .bootstrapServers(bootstrapServers())
@@ -281,14 +281,6 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
     };
   }
 
-  private static void sleep(int seconds) {
-    try {
-      TimeUnit.SECONDS.sleep(seconds);
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
   @Test
   public void testGetAndChoose() {
     var topic = "test";
diff --git a/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java b/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
index 62593034..3f5e5193 100644
--- a/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
+++ b/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
@@ -16,15 +16,16 @@
  */
 package org.astraea.app.performance;
 
-import java.util.concurrent.TimeUnit;
+import java.time.Duration;
 import org.astraea.app.common.DataUnit;
+import org.astraea.app.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class DataSupplierTest {
 
   @Test
-  void testDuration() throws InterruptedException {
+  void testDuration() {
     var dataSupplier =
         DataSupplier.of(
             ExeTime.of("2s"),
@@ -33,7 +34,7 @@ public class DataSupplierTest {
             DistributionType.FIXED.create(10),
             DataUnit.KiB.of(100));
     Assertions.assertTrue(dataSupplier.get().hasData());
-    TimeUnit.SECONDS.sleep(3);
+    Utils.sleep(Duration.ofSeconds(3));
     Assertions.assertFalse(dataSupplier.get().hasData());
   }
 
diff --git a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index 1b2efb92..6036e09d 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -18,15 +18,16 @@ package org.astraea.app.performance;
 
 import com.beust.jcommander.ParameterException;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.argument.Argument;
+import org.astraea.app.common.Utils;
 import org.astraea.app.concurrent.Executor;
 import org.astraea.app.concurrent.State;
 import org.astraea.app.consumer.Consumer;
@@ -84,7 +85,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(1).create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       admin.offsets(Set.of(topic)).values().forEach(o -> Assertions.assertEquals(0, o.latest()));
 
       Assertions.assertEquals(State.RUNNING, producerExecutors.get(0).execute());
diff --git a/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java b/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java
index 0b93d38e..baf399d5 100644
--- a/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java
+++ b/app/src/test/java/org/astraea/app/performance/ProducerExecutorTest.java
@@ -17,10 +17,10 @@
 package org.astraea.app.performance;
 
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
@@ -46,14 +46,14 @@ public class ProducerExecutorTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(executor.topic()).numberOfPartitions(specifiedPartition + 1).create();
       // wait for topic creation
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       admin
           .offsets(Set.of(executor.topic()))
           .values()
           .forEach(o -> Assertions.assertEquals(0, o.latest()));
       Assertions.assertEquals(State.RUNNING, executor.execute());
       // wait for syncing data
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       // only specified partition gets value
       // for normal producer, there is only one record
       // for transactional producer, the size of transaction is 10 and there is one transaction
@@ -90,7 +90,7 @@ public class ProducerExecutorTest extends RequireBrokerCluster {
   void testObserver(ProducerExecutor executor) throws InterruptedException {
     Assertions.assertEquals(State.RUNNING, executor.execute());
     // wait for async call
-    TimeUnit.SECONDS.sleep(2);
+    Utils.sleep(Duration.ofSeconds(2));
     Assertions.assertEquals(
         executor.transactional() ? 10 : 1, ((Observer) executor.observer()).recordsHook.size());
     Assertions.assertEquals(
diff --git a/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java b/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
index 9450013d..671cc668 100644
--- a/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BeanHandlerTest.java
@@ -16,10 +16,10 @@
  */
 package org.astraea.app.web;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -45,11 +45,11 @@ public class BeanHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testBeans() throws InterruptedException {
+  void testBeans() {
     var topic = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(10).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       var handler = new BeanHandler(admin, name -> jmxServiceURL().getPort());
       var response =
           Assertions.assertInstanceOf(
diff --git a/app/src/test/java/org/astraea/app/web/BrokerHandlerTest.java b/app/src/test/java/org/astraea/app/web/BrokerHandlerTest.java
index 35a405d0..15018e36 100644
--- a/app/src/test/java/org/astraea/app/web/BrokerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BrokerHandlerTest.java
@@ -16,11 +16,11 @@
  */
 package org.astraea.app.web;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -30,11 +30,11 @@ import org.junit.jupiter.api.Test;
 public class BrokerHandlerTest extends RequireBrokerCluster {
 
   @Test
-  void testListBrokers() throws InterruptedException {
+  void testListBrokers() {
     var topic = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(10).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       var handler = new BrokerHandler(admin);
       var response =
           Assertions.assertInstanceOf(
@@ -69,12 +69,12 @@ public class BrokerHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testQuerySingleBroker() throws InterruptedException {
+  void testQuerySingleBroker() {
     var topic = Utils.randomString(10);
     var brokerId = brokerIds().iterator().next();
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(10).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       var handler = new BrokerHandler(admin);
       var broker =
           Assertions.assertInstanceOf(
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 578c914e..4294f296 100644
--- a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
@@ -21,7 +21,6 @@ import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
@@ -33,12 +32,12 @@ import org.junit.jupiter.api.Test;
 public class GroupHandlerTest extends RequireBrokerCluster {
 
   @Test
-  void testListGroups() throws InterruptedException {
+  void testListGroups() {
     var topicName = Utils.randomString(10);
     var groupId = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       try (var consumer =
           Consumer.forTopics(Set.of(topicName))
@@ -111,7 +110,7 @@ public class GroupHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testSpecifyTopic() throws InterruptedException {
+  void testSpecifyTopic() {
     var topicName0 = Utils.randomString(10);
     var topicName1 = Utils.randomString(10);
     var groupId0 = Utils.randomString(10);
@@ -132,7 +131,7 @@ public class GroupHandlerTest extends RequireBrokerCluster {
         Assertions.assertEquals(1, consumer0.poll(1, Duration.ofSeconds(10)).size());
         Assertions.assertEquals(0, consumer1.poll(Duration.ofSeconds(2)).size());
 
-        TimeUnit.SECONDS.sleep(3);
+        Utils.sleep(Duration.ofSeconds(3));
 
         // query for only topicName0 so only group of consumer0 can be returned.
         consumer0.commitOffsets(Duration.ofSeconds(3));
diff --git a/app/src/test/java/org/astraea/app/web/PipelineHandlerTest.java b/app/src/test/java/org/astraea/app/web/PipelineHandlerTest.java
index dee010c5..cb6b3df0 100644
--- a/app/src/test/java/org/astraea/app/web/PipelineHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/PipelineHandlerTest.java
@@ -16,10 +16,10 @@
  */
 package org.astraea.app.web;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.producer.Producer;
@@ -54,7 +54,7 @@ public class PipelineHandlerTest extends RequireBrokerCluster {
     var topic = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(1).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       try (var producer = Producer.of(bootstrapServers())) {
         producer.sender().topic(topic).value(new byte[10]).run().toCompletableFuture().get();
         var handler = new PipelineHandler(admin);
diff --git a/app/src/test/java/org/astraea/app/web/PostRequestTest.java b/app/src/test/java/org/astraea/app/web/PostRequestTest.java
index 1e589b3f..811854bb 100644
--- a/app/src/test/java/org/astraea/app/web/PostRequestTest.java
+++ b/app/src/test/java/org/astraea/app/web/PostRequestTest.java
@@ -20,6 +20,9 @@ import com.sun.net.httpserver.HttpExchange;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -44,6 +47,14 @@ public class PostRequestTest {
     Assertions.assertEquals("10", PostRequest.handleDouble(10.00));
     Assertions.assertEquals("10.01", PostRequest.handleDouble(10.01));
     Assertions.assertEquals("xxxx", PostRequest.handleDouble("xxxx"));
+    Assertions.assertEquals("{\"foo\":1}", PostRequest.handleDouble(Map.of("foo", 1)));
+    Assertions.assertEquals("[\"a\",1]", PostRequest.handleDouble(List.of("a", 1)));
+    Assertions.assertEquals(
+        "[{\"foo\":{\"bar\":10}}]",
+        PostRequest.handleDouble(List.of(Map.of("foo", Map.of("bar", 10)))));
+    Assertions.assertEquals(
+        "[{\"foo\":{\"bar\":1.1}}]",
+        PostRequest.handleDouble(List.of(Map.of("foo", Map.of("bar", 1.1d)))));
   }
 
   @Test
@@ -81,4 +92,53 @@ public class PostRequestTest {
     Assertions.assertEquals(1, request.ints("c").get(0));
     Assertions.assertEquals(2, request.ints("c").get(1));
   }
+
+  @Test
+  void testValues() throws IOException {
+    var input =
+        new ByteArrayInputStream(
+            "{\"a\":[{\"foo\": \"r1\", \"bar\": 1},{\"foo\": \"r2\", \"bar\": 2}]}"
+                .getBytes(StandardCharsets.UTF_8));
+    var exchange = Mockito.mock(HttpExchange.class);
+    Mockito.when(exchange.getRequestBody()).thenReturn(input);
+
+    var request = PostRequest.of(exchange);
+    Assertions.assertEquals(
+        List.of(new ForTestValue("r1", 1), new ForTestValue("r2", 2)),
+        request.values("a", ForTestValue.class));
+  }
+
+  @Test
+  void testValue() throws IOException {
+    var input =
+        new ByteArrayInputStream(
+            "{\"a\":{\"foo\": \"r1\", \"bar\": 1}}".getBytes(StandardCharsets.UTF_8));
+    var exchange = Mockito.mock(HttpExchange.class);
+    Mockito.when(exchange.getRequestBody()).thenReturn(input);
+
+    var request = PostRequest.of(exchange);
+    Assertions.assertEquals(new ForTestValue("r1", 1), request.value("a", ForTestValue.class));
+  }
+
+  class ForTestValue {
+    final String foo;
+    final Integer bar;
+
+    ForTestValue(String foo, Integer bar) {
+      this.foo = foo;
+      this.bar = bar;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if ((obj == null) || (getClass() != obj.getClass())) {
+        return false;
+      }
+      ForTestValue other = (ForTestValue) obj;
+      return Objects.equals(foo, other.foo) && Objects.equals(bar, other.bar);
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/ProducerHandlerTest.java b/app/src/test/java/org/astraea/app/web/ProducerHandlerTest.java
index 8ec21908..b382d4c1 100644
--- a/app/src/test/java/org/astraea/app/web/ProducerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/ProducerHandlerTest.java
@@ -16,11 +16,11 @@
  */
 package org.astraea.app.web;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
@@ -60,7 +60,7 @@ public class ProducerHandlerTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(2).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       var handler = new ProducerHandler(admin);
       producer
           .sender()
@@ -78,7 +78,7 @@ public class ProducerHandlerTest extends RequireBrokerCluster {
           .run()
           .toCompletableFuture()
           .get();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       Assertions.assertEquals(
           1,
@@ -114,7 +114,7 @@ public class ProducerHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testPartitions() throws InterruptedException {
+  void testPartitions() {
     var topicName = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers())) {
       var handler = new ProducerHandler(admin);
@@ -140,7 +140,7 @@ public class ProducerHandlerTest extends RequireBrokerCluster {
                       "a")));
 
       admin.creator().topic(topicName).numberOfPartitions(3).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       Assertions.assertEquals(
           3, handler.partitions(Map.of(ProducerHandler.TOPIC_KEY, topicName)).size());
diff --git a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
index 7b01f5a6..4aecc180 100644
--- a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
@@ -16,10 +16,10 @@
  */
 package org.astraea.app.web;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
@@ -30,12 +30,12 @@ import org.junit.jupiter.api.Test;
 public class ReassignmentHandlerTest extends RequireBrokerCluster {
 
   @Test
-  void testMigrateToAnotherBroker() throws InterruptedException {
+  void testMigrateToAnotherBroker() {
     var topicName = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new ReassignmentHandler(admin);
       admin.creator().topic(topicName).numberOfPartitions(1).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       var currentBroker =
           admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0).broker();
@@ -53,7 +53,7 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
                       ReassignmentHandler.TO_KEY,
                       "[\"" + nextBroker + "\"]"))));
 
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       var reassignments = handler.get(Optional.of(topicName), Map.of());
       // the reassignment should be completed
       Assertions.assertEquals(0, reassignments.reassignments.size());
@@ -65,12 +65,12 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testMigrateToAnotherPath() throws InterruptedException {
+  void testMigrateToAnotherPath() {
     var topicName = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new ReassignmentHandler(admin);
       admin.creator().topic(topicName).numberOfPartitions(1).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
 
       var currentReplica =
           admin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).get(0);
@@ -96,7 +96,7 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
                       ReassignmentHandler.TO_KEY,
                       nextPath))));
 
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
       var reassignments = handler.get(Optional.of(topicName), Map.of());
       // the reassignment should be completed
       Assertions.assertEquals(0, reassignments.reassignments.size());
diff --git a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
index 6faf8634..f206b2f6 100644
--- a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -21,34 +21,32 @@ import static java.util.stream.Collectors.toList;
 import static org.astraea.app.web.RecordHandler.ASYNC;
 import static org.astraea.app.web.RecordHandler.DISTANCE_FROM_BEGINNING;
 import static org.astraea.app.web.RecordHandler.DISTANCE_FROM_LATEST;
-import static org.astraea.app.web.RecordHandler.KEY;
 import static org.astraea.app.web.RecordHandler.KEY_DESERIALIZER;
-import static org.astraea.app.web.RecordHandler.KEY_SERIALIZER;
+import static org.astraea.app.web.RecordHandler.LIMIT;
 import static org.astraea.app.web.RecordHandler.PARTITION;
 import static org.astraea.app.web.RecordHandler.RECORDS;
 import static org.astraea.app.web.RecordHandler.SEEK_TO;
 import static org.astraea.app.web.RecordHandler.TIMEOUT;
-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.TRANSACTION_ID;
 import static org.astraea.app.web.RecordHandler.VALUE_DESERIALIZER;
-import static org.astraea.app.web.RecordHandler.VALUE_SERIALIZER;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 
+import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.Base64;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
+import org.astraea.app.consumer.Deserializer;
 import org.astraea.app.consumer.Header;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -59,6 +57,7 @@ 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;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public class RecordHandlerTest extends RequireBrokerCluster {
 
@@ -67,48 +66,103 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var handler = new RecordHandler(bootstrapServers());
     Assertions.assertThrows(
         IllegalArgumentException.class,
-        () -> handler.post(PostRequest.of(Map.of())),
+        () -> handler.post(PostRequest.of(Map.of(RECORDS, "[]"))),
+        "records should contain at least one record");
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> handler.post(PostRequest.of(Map.of(RECORDS, "[{}]"))),
         "topic must be set");
   }
 
   @Test
-  void testPost() {
+  void testPostTimeout() {
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> new RecordHandler(bootstrapServers()).post(PostRequest.of(Map.of(TIMEOUT, "foo"))));
+    Assertions.assertInstanceOf(
+        RecordHandler.PostResponse.class,
+        new RecordHandler(bootstrapServers())
+            .post(
+                PostRequest.of(
+                    new Gson()
+                        .toJson(
+                            Map.of(
+                                TIMEOUT,
+                                "10s",
+                                RECORDS,
+                                List.of(
+                                    new RecordHandler.PostRecord(
+                                        "test", null, null, null, null, null, null)))))));
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  void testPost(boolean isTransaction) {
     var topic = Utils.randomString(10);
-    var handler = new RecordHandler(bootstrapServers());
     var currentTimestamp = System.currentTimeMillis();
-    var result =
+    var requestParams = new HashMap<String, Object>();
+    requestParams.put(
+        RECORDS,
+        List.of(
+            new RecordHandler.PostRecord(
+                topic, 0, "string", "integer", "foo", 100, currentTimestamp),
+            new RecordHandler.PostRecord(
+                topic, 0, "string", "integer", "bar", 200, currentTimestamp)));
+    if (isTransaction) {
+      requestParams.put(TRANSACTION_ID, "trx-" + topic);
+    }
+    var response =
         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);
+            RecordHandler.PostResponse.class,
+            new RecordHandler(bootstrapServers())
+                .post(PostRequest.of(new Gson().toJson(requestParams))));
+
+    Assertions.assertEquals(2, response.results.size());
+
+    var results = response.results;
+    var metadata = (Metadata) results.get(0);
+    Assertions.assertEquals(0, metadata.offset);
+    Assertions.assertEquals(0, metadata.partition);
+    Assertions.assertEquals(topic, metadata.topic);
+    Assertions.assertEquals("foo".getBytes(UTF_8).length, metadata.serializedKeySize);
+    Assertions.assertEquals(4, metadata.serializedValueSize);
+
+    metadata = (Metadata) results.get(1);
+    Assertions.assertEquals(1, metadata.offset);
+    Assertions.assertEquals(0, metadata.partition);
+    Assertions.assertEquals(topic, metadata.topic);
+    Assertions.assertEquals("bar".getBytes(UTF_8).length, metadata.serializedKeySize);
+    Assertions.assertEquals(4, metadata.serializedValueSize);
 
     try (var consumer =
         Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
             .fromBeginning()
+            .keyDeserializer(Deserializer.STRING)
+            .valueDeserializer(Deserializer.INTEGER)
             .build()) {
-      var record = consumer.poll(1, Duration.ofSeconds(10)).iterator().next();
+      var records = List.copyOf(consumer.poll(2, Duration.ofSeconds(10)));
+      Assertions.assertEquals(2, records.size());
+
+      var record = records.get(0);
       Assertions.assertEquals(topic, record.topic());
       Assertions.assertEquals(currentTimestamp, record.timestamp());
       Assertions.assertEquals(0, record.partition());
+      Assertions.assertEquals(0, record.offset());
       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());
+      Assertions.assertEquals("foo", record.key());
+      Assertions.assertEquals(100, record.value());
+
+      record = records.get(1);
+      Assertions.assertEquals(topic, record.topic());
+      Assertions.assertEquals(currentTimestamp, record.timestamp());
+      Assertions.assertEquals(0, record.partition());
+      Assertions.assertEquals(1, record.offset());
+      Assertions.assertEquals("bar".getBytes(UTF_8).length, record.serializedKeySize());
+      Assertions.assertEquals(4, record.serializedValueSize());
+      Assertions.assertEquals("bar", record.key());
+      Assertions.assertEquals(200, record.value());
     }
   }
 
@@ -122,15 +176,21 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             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"))));
+                    new Gson()
+                        .toJson(
+                            Map.of(
+                                ASYNC,
+                                "true",
+                                RECORDS,
+                                List.of(
+                                    new RecordHandler.PostRecord(
+                                        topic,
+                                        0,
+                                        "string",
+                                        "integer",
+                                        "foo",
+                                        "100",
+                                        currentTimestamp)))))));
     Assertions.assertEquals(Response.ACCEPT, result);
 
     handler.producer.flush();
@@ -157,9 +217,16 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var topic = Utils.randomString(10);
     var handler = new RecordHandler(bootstrapServers());
     Assertions.assertInstanceOf(
-        Metadata.class,
+        RecordHandler.PostResponse.class,
         handler.post(
-            PostRequest.of(Map.of(KEY_SERIALIZER, serializer, TOPIC, topic, KEY, actual))));
+            PostRequest.of(
+                new Gson()
+                    .toJson(
+                        Map.of(
+                            RECORDS,
+                            List.of(
+                                new RecordHandler.PostRecord(
+                                    topic, null, serializer, null, actual, null, null)))))));
 
     try (var consumer =
         Consumer.forTopics(Set.of(topic))
@@ -274,13 +341,13 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testGetRecordByPartition() throws InterruptedException {
+  void testGetRecordByPartition() {
     var topic = Utils.randomString(10);
     try (var admin = Admin.of(bootstrapServers());
         var producer = Producer.of(bootstrapServers())) {
       var partitionNum = 2;
       admin.creator().topic(topic).numberOfPartitions(partitionNum).create();
-      TimeUnit.SECONDS.sleep(2);
+      Utils.sleep(Duration.ofSeconds(2));
 
       for (int partitionId = 0; partitionId < partitionNum; partitionId++) {
         for (int recordIdx = 0; recordIdx < 10; recordIdx++) {
@@ -306,7 +373,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testRecordNum() {
+  void testLimit() {
     var topic = Utils.randomString(10);
     produceData(topic, 10);
 
@@ -316,7 +383,7 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             RecordHandler.Records.class,
             handler.get(
                 Optional.of(topic),
-                Map.of(DISTANCE_FROM_BEGINNING, "2", RECORDS, "3", VALUE_DESERIALIZER, "integer")));
+                Map.of(DISTANCE_FROM_BEGINNING, "2", LIMIT, "3", VALUE_DESERIALIZER, "integer")));
 
     Assertions.assertEquals(3, response.data.size());
     Assertions.assertEquals(
@@ -492,17 +559,22 @@ public class RecordHandlerTest extends RequireBrokerCluster {
     var handler = new RecordHandler(bootstrapServers());
     var currentTimestamp = System.currentTimeMillis();
     Assertions.assertInstanceOf(
-        Metadata.class,
+        RecordHandler.PostResponse.class,
         handler.post(
             PostRequest.of(
-                Map.of(
-                    KEY_SERIALIZER, "string",
-                    VALUE_SERIALIZER, "integer",
-                    TOPIC, topic,
-                    KEY, "foo",
-                    VALUE, "100",
-                    TIMESTAMP, "" + currentTimestamp,
-                    PARTITION, "0"))));
+                new Gson()
+                    .toJson(
+                        Map.of(
+                            RECORDS,
+                            List.of(
+                                new RecordHandler.PostRecord(
+                                    topic,
+                                    0,
+                                    "string",
+                                    "integer",
+                                    "foo",
+                                    "100",
+                                    currentTimestamp)))))));
 
     var records =
         Assertions.assertInstanceOf(
diff --git a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
index c1d91ef6..10b3d857 100644
--- a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
@@ -16,11 +16,11 @@
  */
 package org.astraea.app.web;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -30,11 +30,11 @@ import org.junit.jupiter.api.Test;
 public class TopicHandlerTest extends RequireBrokerCluster {
 
   @Test
-  void testListTopics() throws InterruptedException {
+  void testListTopics() {
     var topicName = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       var handler = new TopicHandler(admin);
       var response =
           Assertions.assertInstanceOf(
@@ -62,11 +62,11 @@ public class TopicHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testQuerySingleTopic() throws InterruptedException {
+  void testQuerySingleTopic() {
     var topicName = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       var handler = new TopicHandler(admin);
       var topicInfo =
           Assertions.assertInstanceOf(
@@ -77,11 +77,11 @@ public class TopicHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testTopics() throws InterruptedException {
+  void testTopics() {
     var topicName = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
-      TimeUnit.SECONDS.sleep(3);
+      Utils.sleep(Duration.ofSeconds(3));
       var handler = new TopicHandler(admin);
       Assertions.assertEquals(Set.of(topicName), handler.topicNames(Optional.of(topicName)));
       Assertions.assertThrows(
@@ -105,7 +105,7 @@ public class TopicHandlerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testCreateTopicWithReplicas() throws InterruptedException {
+  void testCreateTopicWithReplicas() {
     var topicName = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new TopicHandler(admin);
@@ -127,7 +127,7 @@ public class TopicHandlerTest extends RequireBrokerCluster {
 
       // the topic creation is not synced, so we have to wait the creation.
       if (topicInfo.partitions.isEmpty()) {
-        TimeUnit.SECONDS.sleep(2);
+        Utils.sleep(Duration.ofSeconds(2));
         var result = admin.replicas(Set.of(topicName));
         Assertions.assertEquals(2, result.size());
         result.values().forEach(replicas -> Assertions.assertEquals(2, replicas.size()));