From a17fa0c56a7f9abfffa14ff4214e9440e5cc634d Mon Sep 17 00:00:00 2001
From: Robin Han <hanxvdovehx@gmail.com>
Date: Mon, 21 Aug 2023 17:47:19 +0800
Subject: [PATCH] feat(stream_client): add simple stream fetch

Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
---
 checkstyle/import-control-core.xml            |  2 +
 .../src/main/scala/kafka/log/s3/S3Client.java | 12 +--
 .../src/main/scala/kafka/log/s3/S3Stream.java | 89 ++++++++++++++++---
 .../scala/kafka/log/s3/S3StreamClient.java    | 12 +--
 core/src/main/scala/kafka/log/s3/Wal.java     | 12 +--
 .../kafka/log/s3/cache/ReadDataBlock.java     | 59 ++++++++++++
 .../kafka/log/s3/cache/S3BlockCache.java      | 16 ++--
 .../kafka/log/s3/model/RangeMetadata.java     | 19 ++--
 .../kafka/log/s3/model/StreamMetadata.java    | 12 +--
 .../kafka/log/s3/model/StreamRecordBatch.java | 41 +++++++--
 .../objects/CommitCompactObjectRequest.java   | 12 +--
 .../s3/objects/CommitStreamObjectRequest.java | 12 +--
 .../s3/objects/CommitWalObjectRequest.java    | 18 ++--
 .../s3/objects/CommitWalObjectResponse.java   | 12 +--
 .../kafka/log/s3/objects/ObjectManager.java   | 14 +--
 .../kafka/log/s3/objects/ObjectRange.java     | 54 -----------
 ...treamIndex.java => ObjectStreamRange.java} | 22 +++--
 .../kafka/log/s3/objects/StreamObject.java    | 12 +--
 .../kafka/log/s3/operator/S3Operator.java     | 12 +--
 .../scala/kafka/log/s3/operator/Writer.java   | 12 +--
 .../kafka/log/s3/streams/StreamManager.java   | 20 +++--
 .../java/kafka/log/s3/DefaultRecordBatch.java | 59 ++++++++++++
 .../log/s3/DefaultRecordBatchWithContext.java | 63 +++++++++++++
 .../test/java/kafka/log/s3/S3StreamTest.java  | 80 +++++++++++++++++
 24 files changed, 498 insertions(+), 178 deletions(-)
 create mode 100644 core/src/main/scala/kafka/log/s3/cache/ReadDataBlock.java
 delete mode 100644 core/src/main/scala/kafka/log/s3/objects/ObjectRange.java
 rename core/src/main/scala/kafka/log/s3/objects/{WalObjectStreamIndex.java => ObjectStreamRange.java} (76%)
 create mode 100644 core/src/test/java/kafka/log/s3/DefaultRecordBatch.java
 create mode 100644 core/src/test/java/kafka/log/s3/DefaultRecordBatchWithContext.java
 create mode 100644 core/src/test/java/kafka/log/s3/S3StreamTest.java

diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml
index b91ff4f3e7..ab85406d16 100644
--- a/checkstyle/import-control-core.xml
+++ b/checkstyle/import-control-core.xml
@@ -36,6 +36,7 @@
   <allow pkg="kafka.utils" />
   <allow pkg="kafka.serializer" />
   <allow pkg="org.apache.kafka.common" />
+  <allow pkg="org.mockito" />
 
 
   <!-- see KIP-544 for why KafkaYammerMetrics should be used instead of the global default yammer metrics registry
@@ -50,6 +51,7 @@
     <allow pkg="io.netty" />
     <allow pkg="redis.clients.jedis" />
     <allow pkg="kafka.server" />
+    <allow pkg="kafka.log.s3" />
   </subpackage>
 
   <subpackage name="testkit">
diff --git a/core/src/main/scala/kafka/log/s3/S3Client.java b/core/src/main/scala/kafka/log/s3/S3Client.java
index 35dbacc5cc..f6153344e4 100644
--- a/core/src/main/scala/kafka/log/s3/S3Client.java
+++ b/core/src/main/scala/kafka/log/s3/S3Client.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/S3Stream.java b/core/src/main/scala/kafka/log/s3/S3Stream.java
index 8884c89c79..3a9650d679 100644
--- a/core/src/main/scala/kafka/log/s3/S3Stream.java
+++ b/core/src/main/scala/kafka/log/s3/S3Stream.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -17,19 +17,30 @@
 
 package kafka.log.s3;
 
+import com.automq.elasticstream.client.DefaultAppendResult;
 import com.automq.elasticstream.client.api.AppendResult;
 import com.automq.elasticstream.client.api.FetchResult;
 import com.automq.elasticstream.client.api.RecordBatch;
+import com.automq.elasticstream.client.api.RecordBatchWithContext;
 import com.automq.elasticstream.client.api.Stream;
+import kafka.log.s3.cache.ReadDataBlock;
 import kafka.log.s3.cache.S3BlockCache;
 import kafka.log.s3.model.StreamMetadata;
+import kafka.log.s3.model.StreamRecordBatch;
 import kafka.log.s3.objects.ObjectManager;
 import kafka.log.s3.streams.StreamManager;
 
+import java.util.LinkedList;
+import java.util.List;
+import java.util.OptionalLong;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
 
 public class S3Stream implements Stream {
     private final StreamMetadata metadata;
+    private final long streamId;
+    private final long epoch;
+    private final AtomicLong nextOffset;
     private final Wal wal;
     private final S3BlockCache blockCache;
     private final StreamManager streamManager;
@@ -37,6 +48,9 @@ public class S3Stream implements Stream {
 
     public S3Stream(StreamMetadata metadata, Wal wal, S3BlockCache blockCache, StreamManager streamManager, ObjectManager objectManager) {
         this.metadata = metadata;
+        this.streamId = metadata.getStreamId();
+        this.epoch = metadata.getEpoch();
+        this.nextOffset = new AtomicLong(metadata.getRanges().get(metadata.getRanges().size() - 1).getStartOffset());
         this.wal = wal;
         this.blockCache = blockCache;
         this.streamManager = streamManager;
@@ -55,31 +69,84 @@ public long startOffset() {
 
     @Override
     public long nextOffset() {
-        return 0;
+        return nextOffset.get();
     }
 
     @Override
     public CompletableFuture<AppendResult> append(RecordBatch recordBatch) {
-        return null;
+        long offset = nextOffset.getAndIncrement();
+        StreamRecordBatch streamRecordBatch = new StreamRecordBatch(streamId, epoch, offset, recordBatch);
+        return wal.append(streamRecordBatch).thenApply(nil -> new DefaultAppendResult(offset));
     }
 
     @Override
     public CompletableFuture<FetchResult> fetch(long startOffset, long endOffset, int maxBytes) {
-        return null;
+        //TODO: bound check
+        //TODO: concurrent read.
+        //TODO: async read.
+        List<Long> objects = objectManager.getObjects(streamId, startOffset, endOffset, maxBytes);
+        long nextStartOffset = startOffset;
+        int nextMaxBytes = maxBytes;
+        List<RecordBatchWithContext> records = new LinkedList<>();
+        for (long objectId : objects) {
+            try {
+                ReadDataBlock dataBlock = blockCache.read(objectId, streamId, nextStartOffset, endOffset, nextMaxBytes).get();
+                OptionalLong blockStartOffset = dataBlock.startOffset();
+                if (blockStartOffset.isEmpty()) {
+                    throw new IllegalStateException("expect not empty block from object[" + objectId + "]");
+                }
+                if (blockStartOffset.getAsLong() != nextStartOffset) {
+                    throw new IllegalStateException("records not continuous, expect start offset[" + nextStartOffset + "], actual["
+                            + blockStartOffset.getAsLong() + "]");
+                }
+                records.addAll(dataBlock.getRecords());
+                // Already check start offset, so it's safe to get end offset.
+                //noinspection OptionalGetWithoutIsPresent
+                nextStartOffset = dataBlock.endOffset().getAsLong();
+                nextMaxBytes -= Math.min(nextMaxBytes, dataBlock.sizeInBytes());
+                if (nextStartOffset >= endOffset || nextMaxBytes <= 0) {
+                    break;
+                }
+            } catch (Throwable e) {
+                throw new RuntimeException(e);
+            }
+        }
+        //TODO: records integrity check.
+        return CompletableFuture.completedFuture(new DefaultFetchResult(records));
     }
 
     @Override
     public CompletableFuture<Void> trim(long newStartOffset) {
+        if (newStartOffset < metadata.getStartOffset()) {
+            throw new IllegalArgumentException("newStartOffset[" + newStartOffset + "] cannot be less than current start offset["
+                    + metadata.getStartOffset() + "]");
+        }
+        metadata.setStartOffset(newStartOffset);
         return streamManager.trimStream(metadata.getStreamId(), metadata.getEpoch(), newStartOffset);
     }
 
     @Override
     public CompletableFuture<Void> close() {
-        return null;
+        // TODO: add stream status to fence future access.
+        return CompletableFuture.completedFuture(null);
     }
 
     @Override
     public CompletableFuture<Void> destroy() {
-        return null;
+        // TODO: add stream status to fence future access.
+        return streamManager.deleteStream(streamId, epoch);
+    }
+
+    static class DefaultFetchResult implements FetchResult {
+        private final List<RecordBatchWithContext> records;
+
+        public DefaultFetchResult(List<RecordBatchWithContext> records) {
+            this.records = records;
+        }
+
+        @Override
+        public List<RecordBatchWithContext> recordBatchList() {
+            return records;
+        }
     }
 }
diff --git a/core/src/main/scala/kafka/log/s3/S3StreamClient.java b/core/src/main/scala/kafka/log/s3/S3StreamClient.java
index 4eca190c6e..2c473c0d1d 100644
--- a/core/src/main/scala/kafka/log/s3/S3StreamClient.java
+++ b/core/src/main/scala/kafka/log/s3/S3StreamClient.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/Wal.java b/core/src/main/scala/kafka/log/s3/Wal.java
index 14088e0106..6050cd985e 100644
--- a/core/src/main/scala/kafka/log/s3/Wal.java
+++ b/core/src/main/scala/kafka/log/s3/Wal.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/cache/ReadDataBlock.java b/core/src/main/scala/kafka/log/s3/cache/ReadDataBlock.java
new file mode 100644
index 0000000000..35110b35b7
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/cache/ReadDataBlock.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log.s3.cache;
+
+import com.automq.elasticstream.client.api.RecordBatchWithContext;
+
+import java.util.List;
+import java.util.OptionalLong;
+
+public class ReadDataBlock {
+    private List<RecordBatchWithContext> records;
+
+    public ReadDataBlock(List<RecordBatchWithContext> records) {
+        this.records = records;
+    }
+
+    public List<RecordBatchWithContext> getRecords() {
+        return records;
+    }
+
+    public void setRecords(List<RecordBatchWithContext> records) {
+        this.records = records;
+    }
+
+    public OptionalLong startOffset() {
+        if (records.isEmpty()) {
+            return OptionalLong.empty();
+        } else {
+            return OptionalLong.of(records.get(0).baseOffset());
+        }
+    }
+
+    public OptionalLong endOffset() {
+        if (records.isEmpty()) {
+            return OptionalLong.empty();
+        } else {
+            return OptionalLong.of(records.get(records.size() - 1).lastOffset());
+        }
+    }
+
+    public int sizeInBytes() {
+        return records.stream().mapToInt(r -> r.rawPayload().remaining()).sum();
+    }
+}
diff --git a/core/src/main/scala/kafka/log/s3/cache/S3BlockCache.java b/core/src/main/scala/kafka/log/s3/cache/S3BlockCache.java
index 8a117dc7f0..2d2fdae366 100644
--- a/core/src/main/scala/kafka/log/s3/cache/S3BlockCache.java
+++ b/core/src/main/scala/kafka/log/s3/cache/S3BlockCache.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -17,8 +17,6 @@
 
 package kafka.log.s3.cache;
 
-import com.automq.elasticstream.client.api.RecordBatch;
-
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -28,6 +26,6 @@
  */
 public interface S3BlockCache {
 
-    CompletableFuture<RecordBatch> read(long objectId, long streamId, long startOffset, long endOffset, long maxBytes);
+    CompletableFuture<ReadDataBlock> read(long objectId, long streamId, long startOffset, long endOffset, int maxBytes);
 
 }
diff --git a/core/src/main/scala/kafka/log/s3/model/RangeMetadata.java b/core/src/main/scala/kafka/log/s3/model/RangeMetadata.java
index 74dd963a0f..4f3e0c9992 100644
--- a/core/src/main/scala/kafka/log/s3/model/RangeMetadata.java
+++ b/core/src/main/scala/kafka/log/s3/model/RangeMetadata.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -26,6 +26,13 @@ public class RangeMetadata {
     private long endOffset;
     private long serverId;
 
+    public RangeMetadata(int index, long startOffset, long endOffset, long serverId) {
+        this.index = index;
+        this.startOffset = startOffset;
+        this.endOffset = endOffset;
+        this.serverId = serverId;
+    }
+
     public int getIndex() {
         return index;
     }
diff --git a/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java b/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java
index 7e43c037d4..7be834a100 100644
--- a/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java
+++ b/core/src/main/scala/kafka/log/s3/model/StreamMetadata.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/model/StreamRecordBatch.java b/core/src/main/scala/kafka/log/s3/model/StreamRecordBatch.java
index d08f1ca536..52acd4eee6 100644
--- a/core/src/main/scala/kafka/log/s3/model/StreamRecordBatch.java
+++ b/core/src/main/scala/kafka/log/s3/model/StreamRecordBatch.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -20,8 +20,31 @@
 import com.automq.elasticstream.client.api.RecordBatch;
 
 public class StreamRecordBatch {
-    private long streamId;
+    private final long streamId;
+    private final long epoch;
+    private final long startOffset;
+    private final RecordBatch recordBatch;
 
-    private RecordBatch recordBatch;
-    private long startOffset;
+    public StreamRecordBatch(long streamId, long epoch, long startOffset, RecordBatch recordBatch) {
+        this.streamId = streamId;
+        this.epoch = epoch;
+        this.startOffset = startOffset;
+        this.recordBatch = recordBatch;
+    }
+
+    public long getStreamId() {
+        return streamId;
+    }
+
+    public long getEpoch() {
+        return epoch;
+    }
+
+    public long getStartOffset() {
+        return startOffset;
+    }
+
+    public RecordBatch getRecordBatch() {
+        return recordBatch;
+    }
 }
diff --git a/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java b/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java
index be886e636e..848c80878e 100644
--- a/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java
+++ b/core/src/main/scala/kafka/log/s3/objects/CommitCompactObjectRequest.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/objects/CommitStreamObjectRequest.java b/core/src/main/scala/kafka/log/s3/objects/CommitStreamObjectRequest.java
index 84570ad26e..8f8e6823a3 100644
--- a/core/src/main/scala/kafka/log/s3/objects/CommitStreamObjectRequest.java
+++ b/core/src/main/scala/kafka/log/s3/objects/CommitStreamObjectRequest.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java b/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java
index d4d12ee66c..1f33c04108 100644
--- a/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java
+++ b/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectRequest.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -22,7 +22,7 @@
 public class CommitWalObjectRequest {
     private long objectId;
     private long objectSize;
-    private List<WalObjectStreamIndex> streams;
+    private List<ObjectStreamRange> streams;
 
     public long getObjectId() {
         return objectId;
@@ -40,11 +40,11 @@ public void setObjectSize(long objectSize) {
         this.objectSize = objectSize;
     }
 
-    public List<WalObjectStreamIndex> getStreams() {
+    public List<ObjectStreamRange> getStreams() {
         return streams;
     }
 
-    public void setStreams(List<WalObjectStreamIndex> streams) {
+    public void setStreams(List<ObjectStreamRange> streams) {
         this.streams = streams;
     }
 }
diff --git a/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectResponse.java b/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectResponse.java
index f7a285df81..c8985b7bb0 100644
--- a/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectResponse.java
+++ b/core/src/main/scala/kafka/log/s3/objects/CommitWalObjectResponse.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/objects/ObjectManager.java b/core/src/main/scala/kafka/log/s3/objects/ObjectManager.java
index cbd9ebf3aa..1073a2a449 100644
--- a/core/src/main/scala/kafka/log/s3/objects/ObjectManager.java
+++ b/core/src/main/scala/kafka/log/s3/objects/ObjectManager.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -66,7 +66,7 @@ public interface ObjectManager {
     /**
      * Get objects by stream range.
      */
-    List<ObjectRange> getObjects(long streamId, long startOffset, long endOffset, int maxBytes);
+    List<Long> getObjects(long streamId, long startOffset, long endOffset, int maxBytes);
 
 }
 
diff --git a/core/src/main/scala/kafka/log/s3/objects/ObjectRange.java b/core/src/main/scala/kafka/log/s3/objects/ObjectRange.java
deleted file mode 100644
index cd60e0a74b..0000000000
--- a/core/src/main/scala/kafka/log/s3/objects/ObjectRange.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.log.s3.objects;
-
-public class ObjectRange {
-    private long objectId;
-    private long startPosition;
-    private long endPosition;
-
-    public ObjectRange(long objectId, long startPosition, long endPosition) {
-        this.objectId = objectId;
-        this.startPosition = startPosition;
-        this.endPosition = endPosition;
-    }
-
-    public long getObjectId() {
-        return objectId;
-    }
-
-    public void setObjectId(long objectId) {
-        this.objectId = objectId;
-    }
-
-    public long getStartPosition() {
-        return startPosition;
-    }
-
-    public void setStartPosition(long startPosition) {
-        this.startPosition = startPosition;
-    }
-
-    public long getEndPosition() {
-        return endPosition;
-    }
-
-    public void setEndPosition(long endPosition) {
-        this.endPosition = endPosition;
-    }
-}
diff --git a/core/src/main/scala/kafka/log/s3/objects/WalObjectStreamIndex.java b/core/src/main/scala/kafka/log/s3/objects/ObjectStreamRange.java
similarity index 76%
rename from core/src/main/scala/kafka/log/s3/objects/WalObjectStreamIndex.java
rename to core/src/main/scala/kafka/log/s3/objects/ObjectStreamRange.java
index 5c149eda03..e40d7808f8 100644
--- a/core/src/main/scala/kafka/log/s3/objects/WalObjectStreamIndex.java
+++ b/core/src/main/scala/kafka/log/s3/objects/ObjectStreamRange.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -17,8 +17,9 @@
 
 package kafka.log.s3.objects;
 
-public class WalObjectStreamIndex {
+public class ObjectStreamRange {
     private long streamId;
+    private long epoch;
     private long startOffset;
     private long endOffset;
 
@@ -30,6 +31,13 @@ public void setStreamId(long streamId) {
         this.streamId = streamId;
     }
 
+    public long getEpoch() {
+        return epoch;
+    }
+
+    public void setEpoch(long epoch) {
+        this.epoch = epoch;
+    }
     public long getStartOffset() {
         return startOffset;
     }
diff --git a/core/src/main/scala/kafka/log/s3/objects/StreamObject.java b/core/src/main/scala/kafka/log/s3/objects/StreamObject.java
index 6de0400d90..63d8f37474 100644
--- a/core/src/main/scala/kafka/log/s3/objects/StreamObject.java
+++ b/core/src/main/scala/kafka/log/s3/objects/StreamObject.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/operator/S3Operator.java b/core/src/main/scala/kafka/log/s3/operator/S3Operator.java
index 9a9a6a1b1e..b6e383c46e 100644
--- a/core/src/main/scala/kafka/log/s3/operator/S3Operator.java
+++ b/core/src/main/scala/kafka/log/s3/operator/S3Operator.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/operator/Writer.java b/core/src/main/scala/kafka/log/s3/operator/Writer.java
index bcefec5f0f..b311f5e77e 100644
--- a/core/src/main/scala/kafka/log/s3/operator/Writer.java
+++ b/core/src/main/scala/kafka/log/s3/operator/Writer.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/core/src/main/scala/kafka/log/s3/streams/StreamManager.java b/core/src/main/scala/kafka/log/s3/streams/StreamManager.java
index 483efc7a01..7e2cdf3291 100644
--- a/core/src/main/scala/kafka/log/s3/streams/StreamManager.java
+++ b/core/src/main/scala/kafka/log/s3/streams/StreamManager.java
@@ -1,13 +1,13 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -50,5 +50,13 @@ public interface StreamManager {
      * @param newStartOffset new start offset.
      */
     CompletableFuture<Void> trimStream(long streamId, long epoch, long newStartOffset);
+
+    /**
+     * Delete stream.
+     *
+     * @param streamId stream id.
+     * @param epoch    stream epoch.
+     */
+    CompletableFuture<Void> deleteStream(long streamId, long epoch);
 }
 
diff --git a/core/src/test/java/kafka/log/s3/DefaultRecordBatch.java b/core/src/test/java/kafka/log/s3/DefaultRecordBatch.java
new file mode 100644
index 0000000000..681a1c25c2
--- /dev/null
+++ b/core/src/test/java/kafka/log/s3/DefaultRecordBatch.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log.s3;
+
+import com.automq.elasticstream.client.api.RecordBatch;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+
+public class DefaultRecordBatch implements RecordBatch {
+    int count;
+    ByteBuffer payload;
+
+    public static RecordBatch of(int count, int size) {
+        DefaultRecordBatch record = new DefaultRecordBatch();
+        record.count = count;
+        record.payload = ByteBuffer.allocate(size);
+        record.payload.position(size);
+        record.payload.flip();
+        return record;
+    }
+
+
+    @Override
+    public int count() {
+        return count;
+    }
+
+    @Override
+    public long baseTimestamp() {
+        return 0;
+    }
+
+    @Override
+    public Map<String, String> properties() {
+        return Collections.emptyMap();
+    }
+
+    @Override
+    public ByteBuffer rawPayload() {
+        return payload.duplicate();
+    }
+}
diff --git a/core/src/test/java/kafka/log/s3/DefaultRecordBatchWithContext.java b/core/src/test/java/kafka/log/s3/DefaultRecordBatchWithContext.java
new file mode 100644
index 0000000000..0f4b87c9ac
--- /dev/null
+++ b/core/src/test/java/kafka/log/s3/DefaultRecordBatchWithContext.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log.s3;
+
+import com.automq.elasticstream.client.api.RecordBatch;
+import com.automq.elasticstream.client.api.RecordBatchWithContext;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class DefaultRecordBatchWithContext implements RecordBatchWithContext {
+    private final RecordBatch recordBatch;
+    private final long baseOffset;
+
+    public DefaultRecordBatchWithContext(RecordBatch recordBatch, long baseOffset) {
+        this.recordBatch = recordBatch;
+        this.baseOffset = baseOffset;
+    }
+    @Override
+    public long baseOffset() {
+        return baseOffset;
+    }
+
+    @Override
+    public long lastOffset() {
+        return baseOffset + recordBatch.count();
+    }
+
+    @Override
+    public int count() {
+        return recordBatch.count();
+    }
+
+    @Override
+    public long baseTimestamp() {
+        return recordBatch.baseTimestamp();
+    }
+
+    @Override
+    public Map<String, String> properties() {
+        return recordBatch.properties();
+    }
+
+    @Override
+    public ByteBuffer rawPayload() {
+        return recordBatch.rawPayload();
+    }
+}
diff --git a/core/src/test/java/kafka/log/s3/S3StreamTest.java b/core/src/test/java/kafka/log/s3/S3StreamTest.java
new file mode 100644
index 0000000000..58bdd3bb36
--- /dev/null
+++ b/core/src/test/java/kafka/log/s3/S3StreamTest.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log.s3;
+
+import com.automq.elasticstream.client.api.FetchResult;
+import com.automq.elasticstream.client.api.RecordBatch;
+import com.automq.elasticstream.client.api.RecordBatchWithContext;
+import kafka.log.s3.cache.ReadDataBlock;
+import kafka.log.s3.cache.S3BlockCache;
+import kafka.log.s3.model.RangeMetadata;
+import kafka.log.s3.model.StreamMetadata;
+import kafka.log.s3.objects.ObjectManager;
+import kafka.log.s3.streams.StreamManager;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class S3StreamTest {
+    Wal wal;
+    S3BlockCache blockCache;
+    StreamManager streamManager;
+    ObjectManager objectManager;
+    S3Stream stream;
+
+    @BeforeEach
+    public void setup() {
+        wal = mock(Wal.class);
+        blockCache = mock(S3BlockCache.class);
+        streamManager = mock(StreamManager.class);
+        objectManager = mock(ObjectManager.class);
+        StreamMetadata metadata = new StreamMetadata();
+        metadata.setStreamId(233);
+        metadata.setEpoch(1);
+        metadata.setStartOffset(100);
+        metadata.setRanges(List.of(new RangeMetadata(1, 50, -1, 10)));
+        stream = new S3Stream(metadata, wal, blockCache, streamManager, objectManager);
+    }
+
+    @Test
+    public void testFetch() throws Throwable {
+        when(objectManager.getObjects(eq(233L), eq(110L), eq(120L), eq(100))).thenReturn(List.of(123L, 124L));
+        when(blockCache.read(eq(123L), eq(233L), eq(110L), eq(120L), eq(100)))
+                .thenReturn(CompletableFuture.completedFuture(newReadDataBlock(110, 115, 10)));
+        when(blockCache.read(eq(124L), eq(233L), eq(115L), eq(120L), eq(90)))
+                .thenReturn(CompletableFuture.completedFuture(newReadDataBlock(115, 120, 10)));
+        FetchResult rst = stream.fetch(110, 120, 100).get(1, TimeUnit.SECONDS);
+        assertEquals(2, rst.recordBatchList().size());
+        assertEquals(110, rst.recordBatchList().get(0).baseOffset());
+        assertEquals(120, rst.recordBatchList().get(1).lastOffset());
+    }
+
+    ReadDataBlock newReadDataBlock(long start, long end, int size) {
+        RecordBatch recordBatch = DefaultRecordBatch.of((int) (end - start), size);
+        RecordBatchWithContext recordBatchWithContext = new DefaultRecordBatchWithContext(recordBatch, start);
+        return new ReadDataBlock(List.of(recordBatchWithContext));
+    }
+}