Repository: skiptests/astraea
Pull request #405 (2022-06-12 10:13:19)

Implement topic read web apis
relates to https://github.com/skiptests/astraea/issues/360

brandboat: gentle ping @chia7712 , 再請您撥冗檢視這隻pr，有關 web api doc 的部分，確認api 都沒問題的話，我會另外再開 pr 處理，感謝～

Patch:
diff --git a/app/src/main/java/org/astraea/app/consumer/Builder.java b/app/src/main/java/org/astraea/app/consumer/Builder.java
index 11a13084..6d9f679a 100644
--- a/app/src/main/java/org/astraea/app/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/app/consumer/Builder.java
@@ -16,20 +16,24 @@
  */
 package org.astraea.app.consumer;
 
+import static java.util.Objects.requireNonNull;
+
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Objects;
+import java.util.function.BiConsumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
-abstract class Builder<Key, Value> {
+public abstract class Builder<Key, Value> {
   protected final Map<String, Object> configs = new HashMap<>();
   protected Deserializer<?> keyDeserializer = Deserializer.BYTE_ARRAY;
   protected Deserializer<?> valueDeserializer = Deserializer.BYTE_ARRAY;
-  protected int distanceFromLatest = -1;
+
+  protected SeekStrategy seekStrategy = SeekStrategy.NONE;
+  protected long seekValue = -1;
 
   Builder() {}
 
@@ -54,33 +58,45 @@ abstract class Builder<Key, Value> {
   }
 
   /**
-   * set the offset to read from the latest offset. For example, the end offset is 5, and you set
-   * distanceFromLatest to 2, then you will read data from offset: 3
+   * Set seek strategy name and its accepted value, seek strategies are as follows:
+   * <li>{@link SeekStrategy#DISTANCE_FROM_LATEST DISTANCE_FROM_LATEST}: set the offset to read from
+   *     the latest offset. For example, the end offset is 5, and you set {@code value} to 2,then
+   *     you will read data from offset: 3
+   * <li>{@link SeekStrategy#DISTANCE_FROM_BEGINNING DISTANCE_FROM_BEGINNING}: set the offset to
+   *     read from the beginning offset. For example, the beginning offset is 0, and you set {@code
+   *     value} to 2, then you will read data from offset: 2
+   * <li>{@link SeekStrategy#SEEK_TO SEEK_TO}: set the offset to read from. For example, the {@code
+   *     value} is 15, then you will read data from offset: 15
    *
-   * @param distanceFromLatest the distance from the latest offset
+   * @param seekStrategy seek strategy
+   * @param value the value that will send to seek strategy, if value < 0, throw {@code
+   *     IllegalArgumentException}
    * @return this builder
    */
-  public Builder<Key, Value> distanceFromLatest(int distanceFromLatest) {
-    this.distanceFromLatest = distanceFromLatest;
+  public Builder<Key, Value> seekStrategy(SeekStrategy seekStrategy, long value) {
+    this.seekStrategy = requireNonNull(seekStrategy);
+    if (value < 0) {
+      throw new IllegalArgumentException("seek value should >= 0");
+    }
+    this.seekValue = value;
     return this;
   }
 
   @SuppressWarnings("unchecked")
   public <NewKey> Builder<NewKey, Value> keyDeserializer(Deserializer<NewKey> keyDeserializer) {
-    this.keyDeserializer = Objects.requireNonNull(keyDeserializer);
+    this.keyDeserializer = requireNonNull(keyDeserializer);
     return (Builder<NewKey, Value>) this;
   }
 
   @SuppressWarnings("unchecked")
   public <NewValue> Builder<Key, NewValue> valueDeserializer(
       Deserializer<NewValue> valueDeserializer) {
-    this.valueDeserializer = Objects.requireNonNull(valueDeserializer);
+    this.valueDeserializer = requireNonNull(valueDeserializer);
     return (Builder<Key, NewValue>) this;
   }
 
   public Builder<Key, Value> bootstrapServers(String bootstrapServers) {
-    this.configs.put(
-        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(bootstrapServers));
+    this.configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, requireNonNull(bootstrapServers));
     return this;
   }
 
@@ -121,4 +137,50 @@ abstract class Builder<Key, Value> {
       kafkaConsumer.close();
     }
   }
+
+  public enum SeekStrategy {
+    NONE((consumer, seekValue) -> {}),
+    DISTANCE_FROM_LATEST(
+        (kafkaConsumer, distanceFromLatest) -> {
+          // this mode is not supported by kafka, so we have to calculate the offset first
+          // 1) poll data until the assignment is completed
+          while (kafkaConsumer.assignment().isEmpty()) {
+            kafkaConsumer.poll(Duration.ofMillis(500));
+          }
+          var partitions = kafkaConsumer.assignment();
+          // 2) get the end offsets from all subscribed partitions
+          var endOffsets = kafkaConsumer.endOffsets(partitions);
+          // 3) calculate and then seek to the correct offset (end offset - recent offset)
+          endOffsets.forEach(
+              (tp, latest) -> kafkaConsumer.seek(tp, Math.max(0, latest - distanceFromLatest)));
+        }),
+    DISTANCE_FROM_BEGINNING(
+        (kafkaConsumer, distanceFromBeginning) -> {
+          while (kafkaConsumer.assignment().isEmpty()) {
+            kafkaConsumer.poll(Duration.ofMillis(500));
+          }
+          var partitions = kafkaConsumer.assignment();
+          var beginningOffsets = kafkaConsumer.beginningOffsets(partitions);
+          beginningOffsets.forEach(
+              (tp, beginning) -> kafkaConsumer.seek(tp, beginning + distanceFromBeginning));
+        }),
+    SEEK_TO(
+        (kafkaConsumer, seekTo) -> {
+          while (kafkaConsumer.assignment().isEmpty()) {
+            kafkaConsumer.poll(Duration.ofMillis(500));
+          }
+          var partitions = kafkaConsumer.assignment();
+          partitions.forEach(tp -> kafkaConsumer.seek(tp, seekTo));
+        });
+
+    private final BiConsumer<org.apache.kafka.clients.consumer.Consumer<?, ?>, Long> function;
+
+    SeekStrategy(BiConsumer<org.apache.kafka.clients.consumer.Consumer<?, ?>, Long> function) {
+      this.function = requireNonNull(function);
+    }
+
+    void apply(org.apache.kafka.clients.consumer.Consumer<?, ?> kafkaConsumer, long seekValue) {
+      function.accept(kafkaConsumer, seekValue);
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
index 3c2d5e72..9a267c66 100644
--- a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
@@ -54,19 +54,6 @@ public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
     return this;
   }
 
-  /**
-   * set the offset to read from the latest offset. For example, the end offset is 5, and you set
-   * distanceFromLatest to 2, then you will read data from offset: 3
-   *
-   * @param distanceFromLatest the distance from the latest offset
-   * @return this builder
-   */
-  @Override
-  public PartitionsBuilder<Key, Value> distanceFromLatest(int distanceFromLatest) {
-    super.distanceFromLatest(distanceFromLatest);
-    return this;
-  }
-
   @Override
   public <NewKey> PartitionsBuilder<NewKey, Value> keyDeserializer(
       Deserializer<NewKey> keyDeserializer) {
@@ -111,15 +98,7 @@ public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
             Deserializer.of((Deserializer<Value>) valueDeserializer));
     kafkaConsumer.assign(partitions.stream().map(TopicPartition::to).collect(toList()));
 
-    // this mode is not supported by kafka, so we have to calculate the offset first
-    if (distanceFromLatest > 0) {
-      var partitions = kafkaConsumer.assignment();
-      // 1) get the end offsets from all subscribed partitions
-      var endOffsets = kafkaConsumer.endOffsets(partitions);
-      // 2) calculate and then seek to the correct offset (end offset - recent offset)
-      endOffsets.forEach(
-          (tp, latest) -> kafkaConsumer.seek(tp, Math.max(0, latest - distanceFromLatest)));
-    }
+    seekStrategy.apply(kafkaConsumer, seekValue);
 
     return new AssignedConsumerImpl<>(kafkaConsumer);
   }
diff --git a/app/src/main/java/org/astraea/app/consumer/Record.java b/app/src/main/java/org/astraea/app/consumer/Record.java
index 3c0487e3..53ba0a22 100644
--- a/app/src/main/java/org/astraea/app/consumer/Record.java
+++ b/app/src/main/java/org/astraea/app/consumer/Record.java
@@ -120,4 +120,13 @@ public final class Record<Key, Value> {
   public Collection<Header> headers() {
     return headers;
   }
+
+  /**
+   * Get the leader epoch for the record if available
+   *
+   * @return the leader epoch or empty for legacy record formats
+   */
+  public Optional<Integer> leaderEpoch() {
+    return leaderEpoch;
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
index 052c1e6e..bf54a2be 100644
--- a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -69,19 +69,6 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     return this;
   }
 
-  /**
-   * set the offset to read from the latest offset. For example, the end offset is 5, and you set
-   * distanceFromLatest to 2, then you will read data from offset: 3
-   *
-   * @param distanceFromLatest the distance from the latest offset
-   * @return this builder
-   */
-  @Override
-  public TopicsBuilder<Key, Value> distanceFromLatest(int distanceFromLatest) {
-    super.distanceFromLatest(distanceFromLatest);
-    return this;
-  }
-
   @Override
   public <NewKey> TopicsBuilder<NewKey, Value> keyDeserializer(
       Deserializer<NewKey> keyDeserializer) {
@@ -133,17 +120,7 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
             Deserializer.of((Deserializer<Value>) valueDeserializer));
     kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
 
-    // this mode is not supported by kafka, so we have to calculate the offset first
-    if (distanceFromLatest > 0) {
-      // 1) poll data until the assignment is completed
-      while (kafkaConsumer.assignment().isEmpty()) kafkaConsumer.poll(Duration.ofMillis(500));
-      var partitions = kafkaConsumer.assignment();
-      // 2) get the end offsets from all subscribed partitions
-      var endOffsets = kafkaConsumer.endOffsets(partitions);
-      // 3) calculate and then seek to the correct offset (end offset - recent offset)
-      endOffsets.forEach(
-          (tp, latest) -> kafkaConsumer.seek(tp, Math.max(0, latest - distanceFromLatest)));
-    }
+    seekStrategy.apply(kafkaConsumer, seekValue);
 
     return new SubscribedConsumerImpl<>(kafkaConsumer);
   }
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 163bbe92..c80d51db 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -17,13 +17,31 @@
 package org.astraea.app.web;
 
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
 
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.JsonSerializationContext;
+import com.google.gson.JsonSerializer;
+import java.lang.reflect.Type;
+import java.time.Duration;
 import java.util.Base64;
+import java.util.Collection;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.BiFunction;
+import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
+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.Serializer;
 
@@ -36,36 +54,101 @@ public class RecordHandler implements Handler {
   static final String TIMESTAMP = "timestamp";
   static final String PARTITION = "partition";
   static final String ASYNC = "async";
+  static final String DISTANCE_FROM_LATEST = "distanceFromLatest";
+  static final String DISTANCE_FROM_BEGINNING = "distanceFromBeginning";
+  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 TIMEOUT = "timeout";
 
+  final String bootstrapServers;
   // visible for testing
   final Producer<byte[], byte[]> producer;
 
   RecordHandler(String bootstrapServers) {
-    this.producer = Producer.builder().bootstrapServers(requireNonNull(bootstrapServers)).build();
+    this.bootstrapServers = requireNonNull(bootstrapServers);
+    this.producer = Producer.builder().bootstrapServers(bootstrapServers).build();
   }
 
   @Override
   public Response get(Optional<String> target, Map<String, String> queries) {
-    // TODO: Implement topic read web apis (https://github.com/skiptests/astraea/pull/405)
-    return Response.NOT_FOUND;
+    var topic = target.orElseThrow(() -> new IllegalArgumentException("topic must be set"));
+    var seekStrategies = Set.of(DISTANCE_FROM_LATEST, DISTANCE_FROM_BEGINNING, SEEK_TO);
+    if (queries.keySet().stream().filter(seekStrategies::contains).count() > 1) {
+      throw new IllegalArgumentException("only one seek strategy is allowed");
+    }
+
+    var consumerBuilder =
+        Optional.ofNullable(queries.get(PARTITION))
+            .map(Integer::valueOf)
+            .map(
+                partition ->
+                    (Builder<byte[], byte[]>)
+                        Consumer.forPartitions(Set.of(new TopicPartition(topic, partition))))
+            .orElseGet(() -> Consumer.forTopics(Set.of(topic)));
+
+    var keyDeserializer =
+        Optional.ofNullable(queries.get(KEY_DESERIALIZER))
+            .map(SerDe::of)
+            .orElse(SerDe.STRING)
+            .deserializer;
+    var valueDeserializer =
+        Optional.ofNullable(queries.get(VALUE_DESERIALIZER))
+            .map(SerDe::of)
+            .orElse(SerDe.STRING)
+            .deserializer;
+    consumerBuilder
+        .bootstrapServers(bootstrapServers)
+        .keyDeserializer(keyDeserializer)
+        .valueDeserializer(valueDeserializer);
+
+    Optional.ofNullable(queries.get(DISTANCE_FROM_LATEST))
+        .map(Long::parseLong)
+        .ifPresent(
+            distanceFromLatest ->
+                consumerBuilder.seekStrategy(
+                    Builder.SeekStrategy.DISTANCE_FROM_LATEST, distanceFromLatest));
+
+    Optional.ofNullable(queries.get(DISTANCE_FROM_BEGINNING))
+        .map(Long::parseLong)
+        .ifPresent(
+            distanceFromBeginning ->
+                consumerBuilder.seekStrategy(
+                    Builder.SeekStrategy.DISTANCE_FROM_BEGINNING, distanceFromBeginning));
+
+    Optional.ofNullable(queries.get(SEEK_TO))
+        .map(Long::parseLong)
+        .ifPresent(seekTo -> consumerBuilder.seekStrategy(Builder.SeekStrategy.SEEK_TO, seekTo));
+
+    try (var consumer = consumerBuilder.build()) {
+      var maxRecords = Integer.parseInt(queries.getOrDefault(RECORDS, "1"));
+      return new Records(
+          consumer
+              .poll(
+                  maxRecords,
+                  Duration.ofSeconds(Integer.parseInt(queries.getOrDefault(TIMEOUT, "5"))))
+              .stream()
+              .map(Record::new)
+              // TODO: remove limit here (https://github.com/skiptests/astraea/issues/441)
+              .limit(maxRecords)
+              .collect(toList()));
+    }
   }
 
   @Override
   public Response post(PostRequest request) {
     var topic =
         request.get(TOPIC).orElseThrow(() -> new IllegalArgumentException("topic must be set"));
-    var keySerDe =
-        SerDe.valueOf(
-            request.get(KEY_SERIALIZER).map(String::toUpperCase).orElse(SerDe.STRING.name()));
-    var valueSerDe =
-        SerDe.valueOf(
-            request.get(VALUE_SERIALIZER).map(String::toUpperCase).orElse(SerDe.STRING.name()));
+    var 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(keySerDe.serializer.apply(topic, k)));
-    request.get(VALUE).ifPresent(v -> sender.value(valueSerDe.serializer.apply(topic, v)));
+    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)));
 
@@ -79,41 +162,53 @@ public class RecordHandler implements Handler {
   enum SerDe {
     BYTEARRAY(
         (topic, value) ->
-            Optional.ofNullable(value).map(v -> Base64.getDecoder().decode(v)).orElse(null)),
+            Optional.ofNullable(value).map(v -> Base64.getDecoder().decode(v)).orElse(null),
+        Deserializer.BYTE_ARRAY),
     STRING(
         (topic, value) ->
             Optional.ofNullable(value)
                 .map(v -> Serializer.STRING.serialize(topic, List.of(), value))
-                .orElse(null)),
+                .orElse(null),
+        Deserializer.STRING),
     LONG(
         (topic, value) ->
             Optional.ofNullable(value)
                 .map(Long::parseLong)
                 .map(longVal -> Serializer.LONG.serialize(topic, List.of(), longVal))
-                .orElse(null)),
+                .orElse(null),
+        Deserializer.LONG),
     INTEGER(
         (topic, value) ->
             Optional.ofNullable(value)
                 .map(Integer::parseInt)
                 .map(intVal -> Serializer.INTEGER.serialize(topic, List.of(), intVal))
-                .orElse(null)),
+                .orElse(null),
+        Deserializer.INTEGER),
     FLOAT(
         (topic, value) ->
             Optional.ofNullable(value)
                 .map(Float::parseFloat)
                 .map(floatVal -> Serializer.FLOAT.serialize(topic, List.of(), floatVal))
-                .orElse(null)),
+                .orElse(null),
+        Deserializer.FLOAT),
     DOUBLE(
         (topic, value) ->
             Optional.ofNullable(value)
                 .map(Double::parseDouble)
                 .map(doubleVal -> Serializer.DOUBLE.serialize(topic, List.of(), doubleVal))
-                .orElse(null));
+                .orElse(null),
+        Deserializer.DOUBLE);
 
     final BiFunction<String, String, byte[]> serializer;
+    final Deserializer<?> deserializer;
 
-    SerDe(BiFunction<String, String, byte[]> serializer) {
+    SerDe(BiFunction<String, String, byte[]> serializer, Deserializer<?> deserializer) {
       this.serializer = requireNonNull(serializer);
+      this.deserializer = requireNonNull(deserializer);
+    }
+
+    static SerDe of(String name) {
+      return valueOf(name.toUpperCase(Locale.ROOT));
     }
   }
 
@@ -134,4 +229,70 @@ public class RecordHandler implements Handler {
       serializedValueSize = metadata.serializedValueSize();
     }
   }
+
+  static class Records implements Response {
+    final Collection<Record> data;
+
+    Records(Collection<Record> data) {
+      this.data = data;
+    }
+
+    @Override
+    public String json() {
+      return new GsonBuilder()
+          // gson will do html escape by default (e.g. convert = to \u003d)
+          .disableHtmlEscaping()
+          .registerTypeHierarchyAdapter(byte[].class, new ByteArrayToBase64TypeAdapter())
+          .create()
+          .toJson(this);
+    }
+  }
+
+  static class Record {
+    final String topic;
+    final int partition;
+    final long offset;
+    final long timestamp;
+    final int serializedKeySize;
+    final int serializedValueSize;
+    final Collection<Header> headers;
+    final Object key;
+    final Object value;
+    final Integer leaderEpoch;
+
+    Record(org.astraea.app.consumer.Record<?, ?> record) {
+      topic = record.topic();
+      partition = record.partition();
+      offset = record.offset();
+      timestamp = record.timestamp();
+      serializedKeySize = record.serializedKeySize();
+      serializedValueSize = record.serializedValueSize();
+      headers = record.headers().stream().map(Header::new).collect(toList());
+      key = record.key();
+      value = record.value();
+      leaderEpoch = record.leaderEpoch().orElse(null);
+    }
+  }
+
+  static class Header {
+    final String key;
+    final byte[] value;
+
+    Header(org.astraea.app.consumer.Header header) {
+      this.key = header.key();
+      this.value = header.value();
+    }
+  }
+
+  static class ByteArrayToBase64TypeAdapter
+      implements JsonSerializer<byte[]>, JsonDeserializer<byte[]> {
+    public byte[] deserialize(JsonElement json, Type type, JsonDeserializationContext context)
+        throws JsonParseException {
+      return Base64.getDecoder().decode(json.getAsString());
+    }
+
+    public JsonElement serialize(byte[] src, Type type, JsonSerializationContext context) {
+      return new JsonPrimitive(Base64.getEncoder().encodeToString(src));
+    }
+  }
 }


Test patch:
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 82b74bbf..19b72d66 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -18,6 +18,9 @@ package org.astraea.app.consumer;
 
 import static java.util.Collections.nCopies;
 import static java.util.stream.Collectors.toList;
+import static org.astraea.app.consumer.Builder.SeekStrategy.DISTANCE_FROM_BEGINNING;
+import static org.astraea.app.consumer.Builder.SeekStrategy.DISTANCE_FROM_LATEST;
+import static org.astraea.app.consumer.Builder.SeekStrategy.SEEK_TO;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
@@ -25,6 +28,7 @@ import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.apache.kafka.common.errors.WakeupException;
@@ -172,7 +176,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     try (var consumer =
         Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
-            .distanceFromLatest(3)
+            .seekStrategy(DISTANCE_FROM_LATEST, 3)
             .build()) {
       Assertions.assertEquals(3, consumer.poll(4, Duration.ofSeconds(5)).size());
     }
@@ -180,7 +184,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     try (var consumer =
         Consumer.forTopics(Set.of(topic))
             .bootstrapServers(bootstrapServers())
-            .distanceFromLatest(1000)
+            .seekStrategy(DISTANCE_FROM_LATEST, 1000)
             .build()) {
       Assertions.assertEquals(10, consumer.poll(11, Duration.ofSeconds(5)).size());
     }
@@ -228,7 +232,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     try (var consumer =
         Consumer.forPartitions(Set.of(TopicPartition.of(topic, "1")))
             .bootstrapServers(bootstrapServers())
-            .distanceFromLatest(20)
+            .seekStrategy(DISTANCE_FROM_LATEST, 20)
             .build()) {
       var records = consumer.poll(20, Duration.ofSeconds(5));
       Assertions.assertEquals(10, records.size());
@@ -239,7 +243,7 @@ public class ConsumerTest extends RequireBrokerCluster {
     try (var consumer =
         Consumer.forPartitions(Set.of(TopicPartition.of(topic, "0"), TopicPartition.of(topic, "1")))
             .bootstrapServers(bootstrapServers())
-            .distanceFromLatest(20)
+            .seekStrategy(DISTANCE_FROM_LATEST, 20)
             .build()) {
       var records = consumer.poll(20, Duration.ofSeconds(5));
       Assertions.assertEquals(20, records.size());
@@ -295,4 +299,56 @@ public class ConsumerTest extends RequireBrokerCluster {
       }
     }
   }
+
+  @Test
+  void testDistanceFromBeginning() {
+    var topic = Utils.randomString(10);
+    produceData(topic, 10);
+
+    BiConsumer<Integer, Integer> internalTest =
+        (distanceFromBeginning, expectedSize) -> {
+          try (var consumer =
+              Consumer.forTopics(Set.of(topic))
+                  .bootstrapServers(bootstrapServers())
+                  .seekStrategy(DISTANCE_FROM_BEGINNING, distanceFromBeginning)
+                  .build()) {
+            Assertions.assertEquals(expectedSize, consumer.poll(10, Duration.ofSeconds(5)).size());
+          }
+        };
+
+    internalTest.accept(3, 7);
+    internalTest.accept(1000, 0);
+  }
+
+  @Test
+  void testSeekTo() {
+    var topic = Utils.randomString(10);
+    produceData(topic, 10);
+
+    BiConsumer<Integer, Integer> internalTest =
+        (seekTo, expectedSize) -> {
+          try (var consumer =
+              Consumer.forTopics(Set.of(topic))
+                  .bootstrapServers(bootstrapServers())
+                  .seekStrategy(SEEK_TO, seekTo)
+                  .build()) {
+            Assertions.assertEquals(expectedSize, consumer.poll(10, Duration.ofSeconds(5)).size());
+          }
+        };
+
+    internalTest.accept(9, 1);
+    internalTest.accept(1000, 0);
+  }
+
+  @Test
+  void testInvalidSeekValue() {
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            Consumer.forTopics(Set.of("test"))
+                .bootstrapServers(bootstrapServers())
+                .seekStrategy(SEEK_TO, -1)
+                .build(),
+        "seek value should >= 0");
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
index 69ce157f..fd040f92 100644
--- a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -17,25 +17,41 @@
 package org.astraea.app.web;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+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.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.TIMESTAMP;
 import static org.astraea.app.web.RecordHandler.TOPIC;
 import static org.astraea.app.web.RecordHandler.VALUE;
+import static org.astraea.app.web.RecordHandler.VALUE_DESERIALIZER;
 import static org.astraea.app.web.RecordHandler.VALUE_SERIALIZER;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 
+import com.google.gson.GsonBuilder;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.Base64;
+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.Header;
+import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
+import org.astraea.app.web.RecordHandler.ByteArrayToBase64TypeAdapter;
 import org.astraea.app.web.RecordHandler.Metadata;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -166,4 +182,351 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             Base64.getEncoder().encodeToString("astraea".getBytes(UTF_8)),
             "astraea".getBytes(UTF_8)));
   }
+
+  private static void produceData(String topic, int size) {
+    try (var producer = Producer.builder().bootstrapServers(bootstrapServers()).build()) {
+      IntStream.range(0, size)
+          .forEach(
+              i ->
+                  producer
+                      .sender()
+                      .topic(topic)
+                      .value(ByteBuffer.allocate(Integer.BYTES).putInt(i).array())
+                      .run());
+      producer.flush();
+    }
+  }
+
+  @Test
+  void testInvalidGet() {
+    var handler = new RecordHandler(bootstrapServers());
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> handler.get(Optional.empty(), Map.of()),
+        "topic must be set");
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            handler.get(
+                Optional.of("topic"),
+                Map.of(DISTANCE_FROM_BEGINNING, "1", DISTANCE_FROM_LATEST, "1")),
+        "only one seek strategy is allowed");
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            handler.get(
+                Optional.of("topic"),
+                Map.of(DISTANCE_FROM_BEGINNING, "1", DISTANCE_FROM_LATEST, "1", SEEK_TO, "1")),
+        "only one seek strategy is allowed");
+  }
+
+  @Test
+  void testDistanceFromLatest() {
+    var topic = Utils.randomString(10);
+    produceData(topic, 10);
+
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(DISTANCE_FROM_LATEST, "2", VALUE_DESERIALIZER, "integer")));
+
+    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(
+        List.of(8), response.data.stream().map(record -> record.value).collect(toList()));
+  }
+
+  @Test
+  void testDistanceFromBeginning() {
+    var topic = Utils.randomString(10);
+    produceData(topic, 10);
+
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(DISTANCE_FROM_BEGINNING, "8", VALUE_DESERIALIZER, "integer")));
+
+    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(
+        List.of(8), response.data.stream().map(record -> record.value).collect(toList()));
+  }
+
+  @Test
+  void testSeekTo() {
+    var topic = Utils.randomString(10);
+    produceData(topic, 5);
+
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(Optional.of(topic), Map.of(SEEK_TO, "3", VALUE_DESERIALIZER, "integer")));
+
+    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(
+        List.of(3), response.data.stream().map(record -> record.value).collect(toList()));
+  }
+
+  @Test
+  void testGetRecordByPartition() throws InterruptedException {
+    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);
+
+      for (int partitionId = 0; partitionId < partitionNum; partitionId++) {
+        for (int recordIdx = 0; recordIdx < 10; recordIdx++) {
+          producer
+              .sender()
+              .topic(topic)
+              .partition(partitionId)
+              .value(ByteBuffer.allocate(4).putInt(recordIdx).array())
+              .run();
+        }
+      }
+      producer.flush();
+    }
+
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(Optional.of(topic), Map.of(DISTANCE_FROM_BEGINNING, "1", PARTITION, "1")));
+
+    Assertions.assertEquals(
+        List.of(1), response.data.stream().map(r -> r.partition).collect(toList()));
+  }
+
+  @Test
+  void testRecordNum() {
+    var topic = Utils.randomString(10);
+    produceData(topic, 10);
+
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(DISTANCE_FROM_BEGINNING, "2", RECORDS, "3", VALUE_DESERIALIZER, "integer")));
+
+    Assertions.assertEquals(3, response.data.size());
+    Assertions.assertEquals(
+        List.of(2, 3, 4), response.data.stream().map(record -> record.value).collect(toList()));
+  }
+
+  @ParameterizedTest
+  @MethodSource("forTestDeserializer")
+  void testDeserializer(String valueDeserializer, byte[] value, Object expectedValue) {
+    var topic = Utils.randomString(10);
+    try (var producer = Producer.builder().bootstrapServers(bootstrapServers()).build()) {
+      producer.sender().topic(topic).value(value).run();
+      producer.flush();
+    }
+
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(DISTANCE_FROM_LATEST, "1", VALUE_DESERIALIZER, valueDeserializer)));
+    var records = List.copyOf(response.data);
+    Assertions.assertEquals(1, records.size());
+
+    if (valueDeserializer.equals("bytearray")) {
+      Assertions.assertArrayEquals((byte[]) expectedValue, (byte[]) records.get(0).value);
+    } else {
+      Assertions.assertEquals(expectedValue, records.get(0).value);
+    }
+  }
+
+  private static Stream<Arguments> forTestDeserializer() {
+    return Stream.of(
+        arguments("bytearray", "astraea".getBytes(UTF_8), "astraea".getBytes(UTF_8)),
+        arguments("string", "astraea".getBytes(UTF_8), "astraea"),
+        arguments(
+            "integer",
+            ByteBuffer.allocate(Integer.BYTES).putInt(Integer.MIN_VALUE).array(),
+            Integer.MIN_VALUE),
+        arguments(
+            "float",
+            ByteBuffer.allocate(Float.BYTES).putFloat(Float.MIN_VALUE).array(),
+            Float.MIN_VALUE),
+        arguments(
+            "long",
+            ByteBuffer.allocate(Long.BYTES).putLong(Integer.MIN_VALUE - 1L).array(),
+            Integer.MIN_VALUE - 1L),
+        arguments(
+            "double",
+            ByteBuffer.allocate(Double.BYTES).putDouble(Float.MAX_VALUE - 1d).array(),
+            Float.MAX_VALUE - 1d));
+  }
+
+  @Test
+  void testGetResponse() {
+    var topic = Utils.randomString(10);
+    var timestamp = System.currentTimeMillis();
+    try (var producer = Producer.builder().bootstrapServers(bootstrapServers()).build()) {
+      producer
+          .sender()
+          .topic(topic)
+          .key("astraea".getBytes(UTF_8))
+          .value(ByteBuffer.allocate(Integer.BYTES).putInt(100).array())
+          .headers(List.of(Header.of("a", "b".getBytes(UTF_8))))
+          .timestamp(timestamp)
+          .run();
+      producer.flush();
+    }
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(
+                    DISTANCE_FROM_LATEST,
+                    "1",
+                    KEY_DESERIALIZER,
+                    "string",
+                    VALUE_DESERIALIZER,
+                    "integer")));
+    Assertions.assertEquals(1, response.data.size());
+    var recordDto = response.data.iterator().next();
+    Assertions.assertEquals(topic, recordDto.topic);
+    Assertions.assertEquals(0, recordDto.partition);
+    Assertions.assertEquals(0, recordDto.offset);
+    Assertions.assertEquals(timestamp, recordDto.timestamp);
+    Assertions.assertEquals("astraea", recordDto.key);
+    Assertions.assertEquals(100, recordDto.value);
+    Assertions.assertEquals("astraea".getBytes(UTF_8).length, recordDto.serializedKeySize);
+    Assertions.assertEquals(Integer.BYTES, recordDto.serializedValueSize);
+    Assertions.assertEquals(0, recordDto.leaderEpoch);
+
+    Assertions.assertEquals(1, recordDto.headers.size());
+    var headerDto = recordDto.headers.iterator().next();
+    Assertions.assertEquals("a", headerDto.key);
+    Assertions.assertArrayEquals("b".getBytes(UTF_8), headerDto.value);
+  }
+
+  @Test
+  void testGetJsonResponse() {
+    var topic = Utils.randomString(10);
+    var timestamp = System.currentTimeMillis();
+    try (var producer = Producer.builder().bootstrapServers(bootstrapServers()).build()) {
+      producer
+          .sender()
+          .topic(topic)
+          .key("astraea".getBytes())
+          .value(ByteBuffer.allocate(Integer.BYTES).putInt(100).array())
+          .headers(List.of(Header.of("a", null)))
+          .timestamp(timestamp)
+          .run();
+      producer.flush();
+    }
+    var handler = new RecordHandler(bootstrapServers());
+    var response =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(
+                    DISTANCE_FROM_LATEST,
+                    "1",
+                    KEY_DESERIALIZER,
+                    "bytearray",
+                    VALUE_DESERIALIZER,
+                    "integer")));
+
+    Assertions.assertEquals(
+        "{\"data\":[{"
+            + "\"topic\":\""
+            + topic
+            + "\","
+            + "\"partition\":0,"
+            + "\"offset\":0,"
+            + "\"timestamp\":"
+            + timestamp
+            + ","
+            + "\"serializedKeySize\":7,"
+            + "\"serializedValueSize\":4,"
+            + "\"headers\":[{\"key\":\"a\"}],"
+            + "\"key\":\""
+            + Base64.getEncoder().encodeToString("astraea".getBytes(UTF_8))
+            + "\","
+            + "\"value\":100,"
+            + "\"leaderEpoch\":0"
+            + "}]}",
+        response.json());
+  }
+
+  @Test
+  void testByteArrayToBase64TypeAdapter() {
+    var foo = new Foo("test".getBytes());
+    var gson =
+        new GsonBuilder()
+            .registerTypeHierarchyAdapter(byte[].class, new ByteArrayToBase64TypeAdapter())
+            .create();
+    Assertions.assertArrayEquals(foo.bar, gson.fromJson(gson.toJson(foo), Foo.class).bar);
+  }
+
+  private static class Foo {
+    final byte[] bar;
+
+    public Foo(byte[] bar) {
+      this.bar = bar;
+    }
+  }
+
+  @Test
+  void testPostAndGet() {
+    var topic = Utils.randomString(10);
+    var handler = new RecordHandler(bootstrapServers());
+    var currentTimestamp = System.currentTimeMillis();
+    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"))));
+
+    var records =
+        Assertions.assertInstanceOf(
+            RecordHandler.Records.class,
+            handler.get(
+                Optional.of(topic),
+                Map.of(
+                    DISTANCE_FROM_LATEST,
+                    "1",
+                    KEY_DESERIALIZER,
+                    "string",
+                    VALUE_DESERIALIZER,
+                    "integer",
+                    PARTITION,
+                    "0")));
+    var record = records.data.iterator().next();
+    Assertions.assertEquals(topic, record.topic);
+    Assertions.assertEquals(0, record.partition);
+    Assertions.assertEquals(0, record.offset);
+    Assertions.assertEquals(0, record.leaderEpoch);
+    Assertions.assertEquals("foo".getBytes().length, record.serializedKeySize);
+    Assertions.assertEquals(4, record.serializedValueSize);
+    Assertions.assertEquals("foo", record.key);
+    Assertions.assertEquals(100, record.value);
+    Assertions.assertEquals(currentTimestamp, record.timestamp);
+    Assertions.assertEquals(List.of(), record.headers);
+  }
 }