diff --git a/build.gradle b/build.gradle
index c4311a0ff1..ec4c24c59a 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1021,6 +1021,8 @@ project(':core') {
     implementation libs.slf4jlog4j
     implementation libs.s3Client
 
+    implementation libs.zstd
+
     compileOnly libs.log4j
 
 
diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml
index dcf50cc2c2..477c0a09af 100644
--- a/checkstyle/import-control-core.xml
+++ b/checkstyle/import-control-core.xml
@@ -38,6 +38,7 @@
   <allow pkg="org.apache.kafka.common" />
   <allow pkg="org.mockito" />
   <allow pkg="software.amazon.awssdk" />
+  <allow pkg="com.github.luben.zstd" />
 
 
   <!-- see KIP-544 for why KafkaYammerMetrics should be used instead of the global default yammer metrics registry
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index 727828a514..5eb22ac101 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -1102,10 +1102,10 @@ class Partition(val topicPartition: TopicPartition,
     }
     val lastRecordTimestamp = LAST_RECORD_TIMESTAMP.get()
     val now = System.currentTimeMillis()
-    if (now - lastRecordTimestamp > 1000 && LAST_RECORD_TIMESTAMP.compareAndSet(lastRecordTimestamp, now)) {
+    if (now - lastRecordTimestamp > 10000 && LAST_RECORD_TIMESTAMP.compareAndSet(lastRecordTimestamp, now)) {
       val updateWatermarkStatistics = UPDATE_WATERMARK_TIMER.getAndReset()
       val tryCompleteStatistics = TRY_COMPLETE_TIMER.getAndReset()
-      logger.warn(s"handle offset move cost, maybeIncrementLeaderHW=$updateWatermarkStatistics, tryCompleteDelayedRequests=$tryCompleteStatistics")
+      logger.trace(s"handle offset move cost, maybeIncrementLeaderHW=$updateWatermarkStatistics, tryCompleteDelayedRequests=$tryCompleteStatistics")
     }
   }
   // elastic stream inject end
diff --git a/core/src/main/scala/kafka/log/es/ElasticLog.scala b/core/src/main/scala/kafka/log/es/ElasticLog.scala
index 890bbd6e8b..18488ea3ea 100644
--- a/core/src/main/scala/kafka/log/es/ElasticLog.scala
+++ b/core/src/main/scala/kafka/log/es/ElasticLog.scala
@@ -229,13 +229,13 @@ class ElasticLog(val metaStream: MetaStream,
   private def tryAppendStatistics(): Unit = {
     val lastRecordTimestamp = LAST_RECORD_TIMESTAMP.get()
     val now = System.currentTimeMillis()
-    if (now - lastRecordTimestamp > 1000 && LAST_RECORD_TIMESTAMP.compareAndSet(lastRecordTimestamp, now)) {
+    if (now - lastRecordTimestamp > 10000 && LAST_RECORD_TIMESTAMP.compareAndSet(lastRecordTimestamp, now)) {
       val permitAcquireFailStatistics = APPEND_PERMIT_ACQUIRE_FAIL_TIMER.getAndReset()
       val remainingPermits = APPEND_PERMIT_SEMAPHORE.availablePermits()
       val appendStatistics = APPEND_TIMER.getAndReset()
       val callbackStatistics = APPEND_CALLBACK_TIMER.getAndReset()
       val ackStatistics = APPEND_ACK_TIMER.getAndReset()
-      logger.warn(s"log append cost, permitAcquireFail=$permitAcquireFailStatistics, remainingPermit=$remainingPermits/$APPEND_PERMIT ,append=$appendStatistics, callback=$callbackStatistics, ack=$ackStatistics")
+      logger.info(s"log append cost, permitAcquireFail=$permitAcquireFailStatistics, remainingPermit=$remainingPermits/$APPEND_PERMIT ,append=$appendStatistics, callback=$callbackStatistics, ack=$ackStatistics")
     }
   }
 
diff --git a/core/src/main/scala/kafka/log/es/metrics/Counter.java b/core/src/main/scala/kafka/log/es/metrics/Counter.java
index ff65cbaaff..1488f69f4b 100644
--- a/core/src/main/scala/kafka/log/es/metrics/Counter.java
+++ b/core/src/main/scala/kafka/log/es/metrics/Counter.java
@@ -29,6 +29,10 @@ public void inc() {
         count.add(1);
     }
 
+    public void inc(int n) {
+        count.add(n);
+    }
+
     public Statistics getAndReset() {
         long count = this.count.sumThenReset();
         return new Statistics(count);
diff --git a/core/src/main/scala/kafka/log/s3/ObjectReader.java b/core/src/main/scala/kafka/log/s3/ObjectReader.java
index 90e400dfdb..c60961678a 100644
--- a/core/src/main/scala/kafka/log/s3/ObjectReader.java
+++ b/core/src/main/scala/kafka/log/s3/ObjectReader.java
@@ -21,8 +21,7 @@
 import kafka.log.s3.model.StreamRecordBatch;
 import kafka.log.s3.operator.S3Operator;
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.compress.ZstdFactory;
-import org.apache.kafka.common.utils.BufferSupplier;
+import org.apache.kafka.common.utils.ByteBufferInputStream;
 import org.apache.kafka.common.utils.CloseableIterator;
 import org.apache.kafka.metadata.stream.S3ObjectMetadata;
 import org.slf4j.Logger;
@@ -253,7 +252,11 @@ public DataBlock(ByteBuf buf, int recordCount) {
         public CloseableIterator<StreamRecordBatch> iterator() {
             ByteBuf buf = this.buf.duplicate();
             AtomicInteger remainingRecordCount = new AtomicInteger(recordCount);
-            DataInputStream in = new DataInputStream(ZstdFactory.wrapForInput(buf.nioBuffer(), (byte) 0, BufferSupplier.NO_CACHING));
+            // skip magic and flag
+            buf.skipBytes(2);
+            // TODO: check flag, use uncompressed stream or compressed stream.
+//            DataInputStream in = new DataInputStream(ZstdFactory.wrapForInput(buf.nioBuffer(), (byte) 0, BufferSupplier.NO_CACHING));
+            DataInputStream in = new DataInputStream(new ByteBufferInputStream(buf.nioBuffer()));
             return new CloseableIterator<>() {
                 @Override
                 public boolean hasNext() {
diff --git a/core/src/main/scala/kafka/log/s3/ObjectWriter.java b/core/src/main/scala/kafka/log/s3/ObjectWriter.java
index b978502d0c..dcde77334e 100644
--- a/core/src/main/scala/kafka/log/s3/ObjectWriter.java
+++ b/core/src/main/scala/kafka/log/s3/ObjectWriter.java
@@ -24,18 +24,19 @@
 import kafka.log.s3.objects.ObjectStreamRange;
 import kafka.log.s3.operator.S3Operator;
 import kafka.log.s3.operator.Writer;
-import org.apache.kafka.common.compress.ZstdFactory;
-import org.apache.kafka.common.utils.ByteBufferOutputStream;
 import org.apache.kafka.metadata.stream.ObjectUtils;
 
-import java.io.IOException;
-import java.io.OutputStream;
+import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 // TODO: memory optimization
 public class ObjectWriter {
+
+    private static final byte DATA_BLOCK_MAGIC = 0x01;
+    // TODO: first n bit is the compressed flag
+    private static final byte DATA_BLOCK_DEFAULT_FLAG = 0x02;
     private final int blockSizeThreshold;
     private final int partSizeThreshold;
     private final List<DataBlock> waitingUploadBlocks;
@@ -43,12 +44,9 @@ public class ObjectWriter {
     private IndexBlock indexBlock;
     private final Writer writer;
     private final long objectId;
-    private long nextDataBlockPosition;
 
     private long size;
 
-    private DataBlock dataBlock;
-
     public ObjectWriter(long objectId, S3Operator s3Operator, int blockSizeThreshold, int partSizeThreshold) {
         this.objectId = objectId;
         // TODO: use a better clusterName
@@ -60,78 +58,103 @@ public ObjectWriter(long objectId, S3Operator s3Operator, int blockSizeThreshold
         writer = s3Operator.writer(objectKey);
     }
 
-    public void write(StreamRecordBatch record) {
-        if (dataBlock == null) {
-            dataBlock = new DataBlock(nextDataBlockPosition);
-        }
-        if (dataBlock.write(record)) {
-            waitingUploadBlocks.add(dataBlock);
-            nextDataBlockPosition += dataBlock.size();
-            dataBlock = null;
-            tryUploadPart();
-        }
+    public void write(long streamId, List<StreamRecordBatch> records) {
+        List<List<StreamRecordBatch>> blocks = groupByBlock(records);
+        List<CompletableFuture<Void>> closeCf = new ArrayList<>(blocks.size());
+        blocks.forEach(blockRecords -> {
+            DataBlock block = new DataBlock(streamId, blockRecords);
+            waitingUploadBlocks.add(block);
+            closeCf.add(block.close());
+        });
+        CompletableFuture.allOf(closeCf.toArray(new CompletableFuture[0])).whenComplete((nil, ex) -> tryUploadPart());
+
     }
 
-    public void closeCurrentBlock() {
-        if (dataBlock != null) {
-            dataBlock.close();
-            waitingUploadBlocks.add(dataBlock);
-            nextDataBlockPosition += dataBlock.size();
-            dataBlock = null;
-            tryUploadPart();
+    private List<List<StreamRecordBatch>> groupByBlock(List<StreamRecordBatch> records) {
+        List<List<StreamRecordBatch>> blocks = new LinkedList<>();
+        List<StreamRecordBatch> blockRecords = new ArrayList<>(records.size());
+        for (StreamRecordBatch record : records) {
+            size += record.size();
+            blockRecords.add(record);
+            if (size >= blockSizeThreshold) {
+                blocks.add(blockRecords);
+                blockRecords = new ArrayList<>(records.size());
+                size = 0;
+            }
+        }
+        if (!blockRecords.isEmpty()) {
+            blocks.add(blockRecords);
         }
+        return blocks;
     }
 
-    private void tryUploadPart() {
-        long waitingUploadSize = waitingUploadBlocks.stream().mapToLong(DataBlock::size).sum();
-        if (waitingUploadSize >= partSizeThreshold) {
-            CompositeByteBuf partBuf = Unpooled.compositeBuffer();
+    private synchronized void tryUploadPart() {
+        for (; ; ) {
+            List<DataBlock> uploadBlocks = new ArrayList<>(32);
+            boolean partFull = false;
+            int size = 0;
             for (DataBlock block : waitingUploadBlocks) {
-                partBuf.addComponent(true, block.buffer());
-                completedBlocks.add(block);
+                if (!block.close().isDone()) {
+                    break;
+                }
+                uploadBlocks.add(block);
+                size += block.size();
+                if (size >= partSizeThreshold) {
+                    partFull = true;
+                    break;
+                }
+            }
+            if (partFull) {
+                CompositeByteBuf partBuf = ByteBufAlloc.ALLOC.compositeBuffer();
+                for (DataBlock block : uploadBlocks) {
+                    partBuf.addComponent(true, block.buffer());
+                }
+                writer.write(partBuf);
+                completedBlocks.addAll(uploadBlocks);
+                waitingUploadBlocks.removeIf(uploadBlocks::contains);
+            } else {
+                break;
             }
-            writer.write(partBuf);
-            waitingUploadBlocks.clear();
         }
     }
 
     public CompletableFuture<Void> close() {
-        CompositeByteBuf buf = Unpooled.compositeBuffer();
-        if (dataBlock != null) {
-            dataBlock.close();
-            nextDataBlockPosition += dataBlock.size();
-            waitingUploadBlocks.add(dataBlock);
-            dataBlock = null;
-        }
-        for (DataBlock block : waitingUploadBlocks) {
-            buf.addComponent(true, block.buffer());
-            completedBlocks.add(block);
-        }
-        waitingUploadBlocks.clear();
-        indexBlock = new IndexBlock();
-        buf.addComponent(true, indexBlock.buffer());
-        Footer footer = new Footer(indexBlock.position(), indexBlock.size());
-        buf.addComponent(true, footer.buffer());
-        writer.write(buf.duplicate());
-        size = indexBlock.position() + indexBlock.size() + footer.size();
-        return writer.close();
+        CompletableFuture<Void> waitBlocksCloseCf = CompletableFuture.allOf(waitingUploadBlocks.stream().map(DataBlock::close).toArray(CompletableFuture[]::new));
+        return waitBlocksCloseCf.thenCompose(nil -> {
+            CompositeByteBuf buf = ByteBufAlloc.ALLOC.compositeBuffer();
+            for (DataBlock block : waitingUploadBlocks) {
+                buf.addComponent(true, block.buffer());
+                completedBlocks.add(block);
+            }
+            waitingUploadBlocks.clear();
+            indexBlock = new IndexBlock();
+            buf.addComponent(true, indexBlock.buffer());
+            Footer footer = new Footer(indexBlock.position(), indexBlock.size());
+            buf.addComponent(true, footer.buffer());
+            writer.write(buf.duplicate());
+            size = indexBlock.position() + indexBlock.size() + footer.size();
+            return writer.close();
+        });
     }
 
     public List<ObjectStreamRange> getStreamRanges() {
         List<ObjectStreamRange> streamRanges = new LinkedList<>();
         ObjectStreamRange lastStreamRange = null;
         for (DataBlock block : completedBlocks) {
-            for (ObjectStreamRange streamRange : block.getStreamRanges()) {
-                if (lastStreamRange == null || lastStreamRange.getStreamId() != streamRange.getStreamId()) {
-                    lastStreamRange = new ObjectStreamRange();
-                    lastStreamRange.setStreamId(streamRange.getStreamId());
-                    lastStreamRange.setEpoch(streamRange.getEpoch());
-                    lastStreamRange.setStartOffset(streamRange.getStartOffset());
+            ObjectStreamRange streamRange = block.getStreamRange();
+            if (lastStreamRange == null || lastStreamRange.getStreamId() != streamRange.getStreamId()) {
+                if (lastStreamRange != null) {
                     streamRanges.add(lastStreamRange);
                 }
-                lastStreamRange.setEndOffset(streamRange.getEndOffset());
-
+                lastStreamRange = new ObjectStreamRange();
+                lastStreamRange.setStreamId(streamRange.getStreamId());
+                lastStreamRange.setEpoch(streamRange.getEpoch());
+                lastStreamRange.setStartOffset(streamRange.getStartOffset());
             }
+            lastStreamRange.setEndOffset(streamRange.getEndOffset());
+        }
+        if (lastStreamRange != null) {
+            streamRanges.add(lastStreamRange);
         }
         return streamRanges;
     }
@@ -144,87 +167,41 @@ public long size() {
         return size;
     }
 
-    class DataBlock {
-        private final long position;
-        private ByteBufferOutputStream compressedBlock;
-        private OutputStream out;
-        private ByteBuf compressedBlockBuf;
-        private int blockSize;
-        private final List<ObjectStreamRange> streamRanges;
-        private ObjectStreamRange streamRange;
-        private int recordCount = 0;
-
-        public DataBlock(long position) {
-            this.position = position;
-            compressedBlock = new ByteBufferOutputStream(blockSizeThreshold * 3 / 2);
-            out = ZstdFactory.wrapForOutput(compressedBlock);
-            streamRanges = new LinkedList<>();
-        }
-
-        public boolean write(StreamRecordBatch record) {
-            try {
-                recordCount++;
-                return write0(record);
-            } catch (IOException ex) {
-                // won't happen
-                throw new RuntimeException(ex);
-            }
-        }
-
-        public boolean write0(StreamRecordBatch record) throws IOException {
-            if (streamRange == null || streamRange.getStreamId() != record.getStreamId()) {
-                streamRange = new ObjectStreamRange();
-                streamRange.setStreamId(record.getStreamId());
-                streamRange.setEpoch(record.getEpoch());
-                streamRange.setStartOffset(record.getBaseOffset());
-                streamRanges.add(streamRange);
-            }
-            streamRange.setEndOffset(record.getLastOffset());
-
-            ByteBuf recordBuf = record.encoded();
-            out.write(recordBuf.array(), recordBuf.arrayOffset(), recordBuf.readableBytes());
-            blockSize += recordBuf.readableBytes();
-            if (blockSize >= blockSizeThreshold) {
-                close();
-                return true;
-            }
-            return false;
-        }
-
-        public void close() {
-            try {
-                close0();
-            } catch (IOException e) {
-                throw new RuntimeException(e);
-            }
+    static class DataBlock {
+        private final ByteBuf encodedBuf;
+        private final ObjectStreamRange streamRange;
+        private final int recordCount;
+        private final int size;
+
+        public DataBlock(long streamId, List<StreamRecordBatch> records) {
+            this.streamRange = new ObjectStreamRange(streamId, records.get(0).getEpoch(), records.get(0).getBaseOffset(), records.get(records.size() - 1).getLastOffset());
+            this.recordCount = records.size();
+            int dataSize = records.stream().mapToInt(r -> r.encoded().readableBytes()).sum();
+            this.encodedBuf = ByteBufAlloc.ALLOC.directBuffer(2 + 2 + dataSize);
+            encodedBuf.writeByte(DATA_BLOCK_MAGIC);
+            encodedBuf.writeByte(DATA_BLOCK_DEFAULT_FLAG);
+            records.forEach(r -> encodedBuf.writeBytes(r.encoded()));
+            this.size = encodedBuf.readableBytes();
         }
 
-        private void close0() throws IOException {
-            out.close();
-            compressedBlock.close();
-            compressedBlockBuf = Unpooled.wrappedBuffer(compressedBlock.buffer().duplicate().flip());
-            out = null;
-            compressedBlock = null;
-        }
-
-        public long position() {
-            return position;
+        public CompletableFuture<Void> close() {
+            return CompletableFuture.completedFuture(null);
         }
 
         public int size() {
-            return compressedBlockBuf.readableBytes();
+            return size;
         }
 
         public int recordCount() {
             return recordCount;
         }
 
-        public List<ObjectStreamRange> getStreamRanges() {
-            return streamRanges;
+        public ObjectStreamRange getStreamRange() {
+            return streamRange;
         }
 
         public ByteBuf buffer() {
-            return compressedBlockBuf.duplicate();
+            return encodedBuf.duplicate();
         }
     }
 
@@ -233,31 +210,32 @@ class IndexBlock {
         private final long position;
 
         public IndexBlock() {
-            position = nextDataBlockPosition;
+            long nextPosition = 0;
             buf = Unpooled.buffer(1024 * 1024);
             buf.writeInt(completedBlocks.size()); // block count
             // block index
             for (DataBlock block : completedBlocks) {
                 // start position in the object
-                buf.writeLong(block.position());
+                buf.writeLong(nextPosition);
                 // byte size of the block
                 buf.writeInt(block.size());
                 // how many ranges in the block
                 buf.writeInt(block.recordCount());
+                nextPosition += block.size();
             }
+            position = nextPosition;
             // object stream range
             for (int blockIndex = 0; blockIndex < completedBlocks.size(); blockIndex++) {
                 DataBlock block = completedBlocks.get(blockIndex);
-                for (ObjectStreamRange range : block.getStreamRanges()) {
-                    // stream id of this range
-                    buf.writeLong(range.getStreamId());
-                    // start offset of the related stream
-                    buf.writeLong(range.getStartOffset());
-                    // record count of the related stream in this range
-                    buf.writeInt((int) (range.getEndOffset() - range.getStartOffset()));
-                    // the index of block where this range is in
-                    buf.writeInt(blockIndex);
-                }
+                ObjectStreamRange range = block.getStreamRange();
+                // stream id of this range
+                buf.writeLong(range.getStreamId());
+                // start offset of the related stream
+                buf.writeLong(range.getStartOffset());
+                // record count of the related stream in this range
+                buf.writeInt((int) (range.getEndOffset() - range.getStartOffset()));
+                // the index of block where this range is in
+                buf.writeInt(blockIndex);
             }
         }
 
diff --git a/core/src/main/scala/kafka/log/s3/S3Storage.java b/core/src/main/scala/kafka/log/s3/S3Storage.java
index 1e52fc19b1..13ebbc79d0 100644
--- a/core/src/main/scala/kafka/log/s3/S3Storage.java
+++ b/core/src/main/scala/kafka/log/s3/S3Storage.java
@@ -79,6 +79,21 @@ public void close() {
 
     @Override
     public CompletableFuture<Void> append(StreamRecordBatch streamRecord) {
+        for (;;) {
+            if (logCache.size() < 2L * 1024 * 1024 * 1024) {
+                break;
+            } else {
+                // TODO: log limit
+                LOGGER.warn("log cache size {} is larger than 2GB, wait 100ms", logCache.size());
+                try {
+                    //noinspection BusyWait
+                    Thread.sleep(100);
+                } catch (InterruptedException e) {
+                    break;
+                }
+            }
+        }
+
         WriteAheadLog.AppendResult appendResult = log.append(streamRecord.encoded());
         CompletableFuture<Void> cf = new CompletableFuture<>();
         WalWriteRequest writeRequest = new WalWriteRequest(streamRecord, appendResult.offset, cf);
@@ -191,8 +206,7 @@ private void commitWALObject(WALObjectUploadTaskContext context) {
             walObjectCommitQueue.poll();
             log.trim(context.cache.confirmOffset());
             // transfer records ownership to block cache.
-            blockCache.put(context.cache.records());
-            freeCache(context.cache.blockId());
+            freeCache(context.cache);
             context.cf.complete(null);
 
             // 2. trigger next task to commit.
@@ -203,8 +217,11 @@ private void commitWALObject(WALObjectUploadTaskContext context) {
         }, backgroundExecutor);
     }
 
-    private void freeCache(long blockId) {
-        mainExecutor.execute(() -> logCache.free(blockId));
+    private void freeCache(LogCache.LogCacheBlock cacheBlock) {
+        mainExecutor.execute(() -> {
+            blockCache.put(cacheBlock.records());
+            logCache.free(cacheBlock.blockId());
+        });
     }
 
     static class WALCallbackSequencer {
diff --git a/core/src/main/scala/kafka/log/s3/StreamRecordBatchCodec.java b/core/src/main/scala/kafka/log/s3/StreamRecordBatchCodec.java
index 5d9d575b2e..15047cf1a6 100644
--- a/core/src/main/scala/kafka/log/s3/StreamRecordBatchCodec.java
+++ b/core/src/main/scala/kafka/log/s3/StreamRecordBatchCodec.java
@@ -36,7 +36,7 @@ public static ByteBuf encode(StreamRecordBatch streamRecord) {
                 + 4 // payload length
                 + streamRecord.getRecordBatch().rawPayload().remaining(); // payload
 
-        ByteBuf buf = ByteBufAlloc.ALLOC.heapBuffer(totalLength);
+        ByteBuf buf = ByteBufAlloc.ALLOC.directBuffer(totalLength);
         buf.writeByte(MAGIC_V0);
         buf.writeLong(streamRecord.getStreamId());
         buf.writeLong(streamRecord.getEpoch());
@@ -59,9 +59,10 @@ public static StreamRecordBatch decode(DataInputStream in) {
             long baseOffset = in.readLong();
             int lastOffsetDelta = in.readInt();
             int payloadLength = in.readInt();
-            ByteBuf payload = ByteBufAlloc.ALLOC.heapBuffer(payloadLength);
-            in.readFully(payload.array(), payload.arrayOffset(), payloadLength);
-            payload.writerIndex(payload.readerIndex() + payloadLength);
+            byte[] payloadBytes = new byte[payloadLength];
+            in.readFully(payloadBytes);
+            ByteBuf payload = ByteBufAlloc.ALLOC.directBuffer(payloadLength);
+            payload.writeBytes(payloadBytes);
             return new StreamRecordBatch(streamId, epoch, baseOffset, lastOffsetDelta, payload);
         } catch (IOException e) {
             throw new RuntimeException(e);
diff --git a/core/src/main/scala/kafka/log/s3/WALObjectUploadTask.java b/core/src/main/scala/kafka/log/s3/WALObjectUploadTask.java
index 2e323f3853..a56f9c2b70 100644
--- a/core/src/main/scala/kafka/log/s3/WALObjectUploadTask.java
+++ b/core/src/main/scala/kafka/log/s3/WALObjectUploadTask.java
@@ -87,14 +87,10 @@ private void upload0(long objectId) {
             if (streamSize >= streamSplitSizeThreshold) {
                 streamObjectCfList.add(writeStreamObject(streamRecords));
             } else {
-                for (StreamRecordBatch record : streamRecords) {
-                    walObject.write(record);
-                }
+                walObject.write(streamId, streamRecords);
                 long startOffset = streamRecords.get(0).getBaseOffset();
                 long endOffset = streamRecords.get(streamRecords.size() - 1).getLastOffset();
                 request.addStreamRange(new ObjectStreamRange(streamId, -1L, startOffset, endOffset));
-                // log object block only contain single stream's data.
-                walObject.closeCurrentBlock();
             }
         }
         request.setObjectId(objectId);
@@ -126,10 +122,8 @@ private CompletableFuture<StreamObject> writeStreamObject(List<StreamRecordBatch
         // TODO: retry until success
         return objectIdCf.thenCompose(objectId -> {
             ObjectWriter streamObjectWriter = new ObjectWriter(objectId, s3Operator, objectBlockSize, objectPartSize);
-            for (StreamRecordBatch record : streamRecords) {
-                streamObjectWriter.write(record);
-            }
             long streamId = streamRecords.get(0).getStreamId();
+            streamObjectWriter.write(streamId, streamRecords);
             long startOffset = streamRecords.get(0).getBaseOffset();
             long endOffset = streamRecords.get(streamRecords.size() - 1).getLastOffset();
             StreamObject streamObject = new StreamObject();
diff --git a/core/src/main/scala/kafka/log/s3/ZstdUtils.java b/core/src/main/scala/kafka/log/s3/ZstdUtils.java
new file mode 100644
index 0000000000..6022dfb03d
--- /dev/null
+++ b/core/src/main/scala/kafka/log/s3/ZstdUtils.java
@@ -0,0 +1,45 @@
+/*
+ * 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.github.luben.zstd.Zstd;
+import com.github.luben.zstd.ZstdException;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+
+public class ZstdUtils {
+
+    public static ByteBuf compress(ByteBuf source, ByteBufAllocator alloc) {
+        int uncompressedLength = source.readableBytes();
+        ByteBuf directSource = alloc.directBuffer(uncompressedLength);
+        source.readBytes(directSource);
+        int destLength = uncompressedLength + 1024;
+        for (; ; ) {
+            ByteBuf dest = alloc.directBuffer(destLength, destLength);
+            try {
+                int compressedLength = Zstd.compress(dest.nioBuffer(0, destLength), directSource.nioBuffer());
+                dest.writerIndex(compressedLength);
+                directSource.release();
+                return dest;
+            } catch (ZstdException e) {
+                destLength += 16 * 1024;
+                dest.release();
+            }
+        }
+    }
+}
diff --git a/core/src/main/scala/kafka/log/s3/cache/BlockCache.java b/core/src/main/scala/kafka/log/s3/cache/BlockCache.java
index f5c57d7903..8251078fc5 100644
--- a/core/src/main/scala/kafka/log/s3/cache/BlockCache.java
+++ b/core/src/main/scala/kafka/log/s3/cache/BlockCache.java
@@ -49,6 +49,7 @@ public BlockCache(long maxSize) {
 
     public void put(long streamId, List<StreamRecordBatch> records) {
         if (maxSize == 0 || records.isEmpty()) {
+            records.forEach(StreamRecordBatch::release);
             return;
         }
         boolean overlapped = false;
diff --git a/core/src/main/scala/kafka/log/s3/cache/LogCache.java b/core/src/main/scala/kafka/log/s3/cache/LogCache.java
index 643bac4e2b..6e81be42ee 100644
--- a/core/src/main/scala/kafka/log/s3/cache/LogCache.java
+++ b/core/src/main/scala/kafka/log/s3/cache/LogCache.java
@@ -33,6 +33,7 @@ public class LogCache {
     private final List<LogCacheBlock> archiveBlocks = new ArrayList<>();
     private LogCacheBlock activeBlock;
     private long confirmOffset;
+    private final AtomicLong size = new AtomicLong();
 
     public LogCache(long cacheBlockMaxSize) {
         this.cacheBlockMaxSize = cacheBlockMaxSize;
@@ -40,6 +41,7 @@ public LogCache(long cacheBlockMaxSize) {
     }
 
     public boolean put(StreamRecordBatch recordBatch) {
+        size.addAndGet(recordBatch.size());
         return activeBlock.put(recordBatch);
     }
 
@@ -47,13 +49,13 @@ public boolean put(StreamRecordBatch recordBatch) {
      * Get streamId [startOffset, endOffset) range records with maxBytes limit.
      * If the cache only contain records after startOffset, the return list is empty.
      * Note: the records is retained, the caller should release it.
-     *
      */
     public List<StreamRecordBatch> get(long streamId, long startOffset, long endOffset, int maxBytes) {
         List<StreamRecordBatch> records = get0(streamId, startOffset, endOffset, maxBytes);
         records.forEach(StreamRecordBatch::retain);
         return records;
     }
+
     public List<StreamRecordBatch> get0(long streamId, long startOffset, long endOffset, int maxBytes) {
         List<StreamRecordBatch> rst = new LinkedList<>();
         long nextStartOffset = startOffset;
@@ -93,13 +95,24 @@ public Optional<LogCacheBlock> archiveCurrentBlockIfContains(long streamId) {
     }
 
     public void free(long blockId) {
-        archiveBlocks.removeIf(b -> b.blockId == blockId);
+        archiveBlocks.removeIf(b -> {
+            boolean remove = b.blockId == blockId;
+            if (remove) {
+                size.addAndGet(-b.size);
+                b.records().clear();
+            }
+            return remove;
+        });
     }
 
     public void setConfirmOffset(long confirmOffset) {
         this.confirmOffset = confirmOffset;
     }
 
+    public long size() {
+        return size.get();
+    }
+
     public static class LogCacheBlock {
         private static final AtomicLong BLOCK_ID_ALLOC = new AtomicLong();
         private final long blockId;
diff --git a/core/src/main/scala/kafka/log/s3/metadata/StreamMetadataManager.java b/core/src/main/scala/kafka/log/s3/metadata/StreamMetadataManager.java
index 862ab0d43e..cda835464d 100644
--- a/core/src/main/scala/kafka/log/s3/metadata/StreamMetadataManager.java
+++ b/core/src/main/scala/kafka/log/s3/metadata/StreamMetadataManager.java
@@ -203,15 +203,17 @@ private CompletableFuture<InRangeObjects> fetch0(long streamId, long startOffset
     }
 
     public void retryPendingTasks(List<GetObjectsTask> tasks) {
-        LOGGER.info("[RetryPendingTasks]: retry tasks count: {}", tasks.size());
-        tasks.forEach(task -> {
-            long streamId = task.streamId;
-            long startOffset = task.startOffset;
-            long endOffset = task.endOffset;
-            int limit = task.limit;
-            CompletableFuture<InRangeObjects> newCf = this.fetch(streamId, startOffset, endOffset, limit);
-            FutureUtil.propagate(newCf, task.cf);
-        });
+        if (!tasks.isEmpty()) {
+            LOGGER.info("[RetryPendingTasks]: retry tasks count: {}", tasks.size());
+            tasks.forEach(task -> {
+                long streamId = task.streamId;
+                long startOffset = task.startOffset;
+                long endOffset = task.endOffset;
+                int limit = task.limit;
+                CompletableFuture<InRangeObjects> newCf = this.fetch(streamId, startOffset, endOffset, limit);
+                FutureUtil.propagate(newCf, task.cf);
+            });
+        }
     }
 
     static class GetObjectsTask {
diff --git a/core/src/main/scala/kafka/log/s3/operator/DefaultS3Operator.java b/core/src/main/scala/kafka/log/s3/operator/DefaultS3Operator.java
index 3765678440..9bee9b4618 100644
--- a/core/src/main/scala/kafka/log/s3/operator/DefaultS3Operator.java
+++ b/core/src/main/scala/kafka/log/s3/operator/DefaultS3Operator.java
@@ -19,6 +19,8 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import kafka.log.es.metrics.Counter;
+import kafka.log.es.metrics.Timer;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.common.utils.ThreadUtils;
 import org.slf4j.Logger;
@@ -50,6 +52,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
 public class DefaultS3Operator implements S3Operator {
@@ -58,6 +61,10 @@ public class DefaultS3Operator implements S3Operator {
     public static final String SECRET_KEY_NAME = "KAFKA_S3_SECRET_KEY";
     private final String bucket;
     private final S3AsyncClient s3;
+    private static final Timer PART_UPLOAD_COST = new Timer();
+    private static final Timer OBJECT_UPLOAD_COST = new Timer();
+    private static final Counter OBJECT_UPLOAD_SIZE = new Counter();
+    private static final AtomicLong LAST_LOG_TIMESTAMP = new AtomicLong(System.currentTimeMillis());
     private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
             ThreadUtils.createThreadFactory("s3operator", true));
 
@@ -174,6 +181,7 @@ class DefaultWriter implements Writer {
         private final List<CompletableFuture<CompletedPart>> parts = new LinkedList<>();
         private final AtomicInteger nextPartNumber = new AtomicInteger(1);
         private CompletableFuture<Void> closeCf;
+        private final long start = System.nanoTime();
 
         public DefaultWriter(String path) {
             this.path = path;
@@ -199,10 +207,16 @@ private void init() {
 
         @Override
         public void write(ByteBuf part) {
+            long start = System.nanoTime();
+            OBJECT_UPLOAD_SIZE.inc(part.readableBytes());
             int partNumber = nextPartNumber.getAndIncrement();
             CompletableFuture<CompletedPart> partCf = new CompletableFuture<>();
             parts.add(partCf);
             uploadIdCf.thenAccept(uploadId -> write0(uploadId, partNumber, part, partCf));
+            partCf.whenComplete((nil, ex) -> {
+                PART_UPLOAD_COST.update(System.nanoTime() - start);
+                part.release();
+            });
         }
 
         private void write0(String uploadId, int partNumber, ByteBuf part, CompletableFuture<CompletedPart> partCf) {
@@ -255,13 +269,25 @@ public CompletableFuture<Void> close() {
             if (closeCf != null) {
                 return closeCf;
             }
+            System.out.println("start await close: " + (System.nanoTime() - start) / 1000 / 1000);
             closeCf = new CompletableFuture<>();
             CompletableFuture<Void> uploadDoneCf = uploadIdCf.thenCompose(uploadId -> CompletableFuture.allOf(parts.toArray(new CompletableFuture[0])));
             uploadDoneCf.thenAccept(nil -> {
+                System.out.println("start complete: " + (System.nanoTime() - start) / 1000 / 1000);
                 CompletedMultipartUpload multipartUpload = CompletedMultipartUpload.builder().parts(genCompleteParts()).build();
                 CompleteMultipartUploadRequest request = CompleteMultipartUploadRequest.builder().bucket(bucket).key(path).uploadId(uploadId).multipartUpload(multipartUpload).build();
                 close0(request);
             });
+            closeCf.whenComplete((nil, ex) -> {
+                System.out.println("complete: " + (System.nanoTime() - start) / 1000 / 1000);
+                OBJECT_UPLOAD_COST.update(System.nanoTime() - start);
+                long now = System.currentTimeMillis();
+                if (now - LAST_LOG_TIMESTAMP.get() > 10000) {
+                    LAST_LOG_TIMESTAMP.set(now);
+                    LOGGER.info("upload s3 metrics, object_timer {}, object_size {}, part_timer {}",
+                            OBJECT_UPLOAD_COST.getAndReset(), OBJECT_UPLOAD_SIZE.getAndReset(), PART_UPLOAD_COST.getAndReset());
+                }
+            });
             return closeCf;
         }
 
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index fd2c3583e0..0ab4a52360 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -725,7 +725,7 @@ class KafkaApis(val requestChannel: RequestChannel,
       PRODUCE_ACK_TIMER.update(System.nanoTime() - callbackStartNanos)
       val now = System.currentTimeMillis()
       val lastRecordTimestamp = LAST_RECORD_TIMESTAMP.get();
-      if (now - lastRecordTimestamp > 1000 && LAST_RECORD_TIMESTAMP.compareAndSet(lastRecordTimestamp, now)) {
+      if (now - lastRecordTimestamp > 10000 && LAST_RECORD_TIMESTAMP.compareAndSet(lastRecordTimestamp, now)) {
         val produceStatistics = PRODUCE_TIMER.getAndReset()
         val produceCallbackStatistics = PRODUCE_CALLBACK_TIMER.getAndReset()
         val produceAckStatistics = PRODUCE_ACK_TIMER.getAndReset()
diff --git a/core/src/test/java/kafka/log/es/AlwaysSuccessClientTest.java b/core/src/test/java/kafka/log/es/AlwaysSuccessClientTest.java
index 7c84859df6..926942c131 100644
--- a/core/src/test/java/kafka/log/es/AlwaysSuccessClientTest.java
+++ b/core/src/test/java/kafka/log/es/AlwaysSuccessClientTest.java
@@ -26,6 +26,7 @@
 import com.automq.elasticstream.client.api.RecordBatchWithContext;
 import com.automq.elasticstream.client.api.Stream;
 import com.automq.elasticstream.client.api.StreamClient;
+
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -40,6 +41,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+
 import org.apache.kafka.common.errors.es.SlowFetchHintException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -70,14 +72,14 @@ public void teardown() {
     public void basicAppendAndFetch() throws ExecutionException, InterruptedException {
         client = new AlwaysSuccessClient(new MemoryClient());
         Stream stream = client
-            .streamClient()
-            .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
-            .get();
+                .streamClient()
+                .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
+                .get();
         List<byte[]> payloads = List.of("hello".getBytes(), "world".getBytes());
         CompletableFuture.allOf(
-            payloads
-                .stream()
-                .map(payload -> stream.append(RawPayloadRecordBatch.of(ByteBuffer.wrap(payload)))).toArray(CompletableFuture[]::new)
+                payloads
+                        .stream()
+                        .map(payload -> stream.append(RawPayloadRecordBatch.of(ByteBuffer.wrap(payload)))).toArray(CompletableFuture[]::new)
         ).get();
 
         FetchResult fetched = stream.fetch(0, 100, 1000).get();
@@ -97,17 +99,17 @@ public void testQuickFetch() throws ExecutionException, InterruptedException {
         for (Long delay : quickFetchDelayMillisList) {
             memoryClientWithDelay.setDelayMillis(delay);
             Stream stream = client
-                .streamClient()
-                .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
-                .get();
+                    .streamClient()
+                    .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
+                    .get();
             CompletableFuture.allOf(
-                payloads
-                    .stream()
-                    .map(payload -> stream.append(RawPayloadRecordBatch.of(ByteBuffer.wrap(payload)))).toArray(CompletableFuture[]::new)
+                    payloads
+                            .stream()
+                            .map(payload -> stream.append(RawPayloadRecordBatch.of(ByteBuffer.wrap(payload)))).toArray(CompletableFuture[]::new)
             ).get();
             FetchResult fetched = stream.fetch(0, 100, 1000)
-                .orTimeout(delay + SLOW_FETCH_TIMEOUT_MILLIS / 2, TimeUnit.MILLISECONDS)
-                .get();
+                    .orTimeout(delay + 100, TimeUnit.MILLISECONDS)
+                    .get();
             checkAppendAndFetch(payloads, fetched);
             stream.destroy();
         }
@@ -119,24 +121,24 @@ public void testSlowFetch() throws ExecutionException, InterruptedException {
         client = new AlwaysSuccessClient(memoryClientWithDelay);
         List<byte[]> payloads = List.of("hello".getBytes(), "world".getBytes());
 
-        long slowFetchDelay = SLOW_FETCH_TIMEOUT_MILLIS + SLOW_FETCH_TIMEOUT_MILLIS / 2;
+        long slowFetchDelay = 500 + SLOW_FETCH_TIMEOUT_MILLIS + SLOW_FETCH_TIMEOUT_MILLIS / 2;
         memoryClientWithDelay.setDelayMillis(slowFetchDelay);
         Stream stream = client
-            .streamClient()
-            .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
-            .get();
+                .streamClient()
+                .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
+                .get();
         CompletableFuture.allOf(
-            payloads
-                .stream()
-                .map(payload -> stream.append(RawPayloadRecordBatch.of(ByteBuffer.wrap(payload)))).toArray(CompletableFuture[]::new)
+                payloads
+                        .stream()
+                        .map(payload -> stream.append(RawPayloadRecordBatch.of(ByteBuffer.wrap(payload)))).toArray(CompletableFuture[]::new)
         ).get();
 
         FetchResult fetched = null;
         AtomicBoolean gotSlowFetchHintException = new AtomicBoolean(false);
         try {
             fetched = stream.fetch(0, 100, 1000)
-                .orTimeout(slowFetchDelay + SLOW_FETCH_TIMEOUT_MILLIS / 2, TimeUnit.MILLISECONDS)
-                .get();
+                    .orTimeout(SLOW_FETCH_TIMEOUT_MILLIS * 2, TimeUnit.MILLISECONDS)
+                    .get();
             checkAppendAndFetch(payloads, fetched);
         } catch (ExecutionException e) {
             // should throw SlowFetchHintException after SLOW_FETCH_TIMEOUT_MILLIS ms
@@ -145,8 +147,8 @@ public void testSlowFetch() throws ExecutionException, InterruptedException {
             SeparateSlowAndQuickFetchHint.reset();
             // It should reuse the fetching future above, therefore only (SLOW_FETCH_TIMEOUT_MILLIS / 2) ms is tolerable.
             fetched = stream.fetch(0, 100, 1000)
-                .orTimeout(SLOW_FETCH_TIMEOUT_MILLIS - 1, TimeUnit.MILLISECONDS)
-                .get();
+                    .orTimeout(slowFetchDelay + 100, TimeUnit.MILLISECONDS)
+                    .get();
         }
         checkAppendAndFetch(payloads, fetched);
         assertTrue(gotSlowFetchHintException.get(), "should throw SlowFetchHintException");
@@ -161,12 +163,12 @@ public void testOpenStream() {
 
         AtomicBoolean exceptionThrown = new AtomicBoolean(false);
         openStream(1)
-            .exceptionally(e -> {
-                assertEquals(IOException.class, e.getClass());
-                exceptionThrown.set(true);
-                return null;
-            })
-            .join();
+                .exceptionally(e -> {
+                    assertEquals(IOException.class, e.getClass());
+                    exceptionThrown.set(true);
+                    return null;
+                })
+                .join();
 
         assertTrue(exceptionThrown.get(), "should throw IOException");
     }
@@ -178,40 +180,40 @@ public void testStreamOperationHalt() {
         client = new AlwaysSuccessClient(memoryClientWithDelay);
 
         Stream stream = client
-            .streamClient()
-            .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
-            .join();
+                .streamClient()
+                .createAndOpenStream(CreateStreamOptions.newBuilder().epoch(0).replicaCount(1).build())
+                .join();
 
         AtomicInteger exceptionCount = new AtomicInteger(0);
         stream
-            .append(RawPayloadRecordBatch.of(ByteBuffer.wrap("hello".getBytes())))
-            .exceptionally(e -> {
-                assertEquals(IOException.class, e.getClass());
-                exceptionCount.incrementAndGet();
-                return null;
-            })
-            .join();
+                .append(RawPayloadRecordBatch.of(ByteBuffer.wrap("hello".getBytes())))
+                .exceptionally(e -> {
+                    assertEquals(IOException.class, e.getClass());
+                    exceptionCount.incrementAndGet();
+                    return null;
+                })
+                .join();
         stream.fetch(0, 100, 1000)
-            .exceptionally(e -> {
-                assertEquals(IOException.class, e.getClass());
-                exceptionCount.incrementAndGet();
-                return null;
-            })
-            .join();
+                .exceptionally(e -> {
+                    assertEquals(IOException.class, e.getClass());
+                    exceptionCount.incrementAndGet();
+                    return null;
+                })
+                .join();
         stream.trim(0)
-            .exceptionally(e -> {
-                assertEquals(IOException.class, e.getClass());
-                exceptionCount.incrementAndGet();
-                return null;
-            })
-            .join();
+                .exceptionally(e -> {
+                    assertEquals(IOException.class, e.getClass());
+                    exceptionCount.incrementAndGet();
+                    return null;
+                })
+                .join();
         stream.close()
-            .exceptionally(e -> {
-                assertEquals(IOException.class, e.getClass());
-                exceptionCount.incrementAndGet();
-                return null;
-            })
-            .join();
+                .exceptionally(e -> {
+                    assertEquals(IOException.class, e.getClass());
+                    exceptionCount.incrementAndGet();
+                    return null;
+                })
+                .join();
         assertEquals(4, exceptionCount.get(), "should throw IOException 4 times");
         stream.destroy();
     }
@@ -250,8 +252,8 @@ public void testNormalExceptionHandling() {
 
     private CompletableFuture<Stream> openStream(long streamId) {
         return client
-            .streamClient()
-            .openStream(streamId, OpenStreamOptions.newBuilder().epoch(1).build());
+                .streamClient()
+                .openStream(streamId, OpenStreamOptions.newBuilder().epoch(1).build());
     }
 
     private void checkAppendAndFetch(List<byte[]> rawPayloads, FetchResult fetched) {
@@ -265,6 +267,7 @@ static final class MemoryClientWithDelay extends MemoryClient {
 
         /**
          * Set the additional fetching delay
+         *
          * @param delayMillis
          */
         public void setDelayMillis(long delayMillis) {
@@ -423,9 +426,9 @@ static enum ExceptionHint {
         OK;
 
         private static final List<Exception> OTHER_EXCEPTION_LIST = List.of(
-            new IOException("io exception"),
-            new RuntimeException("runtime exception"),
-            new ElasticStreamClientException(-1, "other exception")
+                new IOException("io exception"),
+                new RuntimeException("runtime exception"),
+                new ElasticStreamClientException(-1, "other exception")
         );
 
         public Exception generateException() {
diff --git a/core/src/test/java/kafka/log/s3/DefaultS3BlockCacheTest.java b/core/src/test/java/kafka/log/s3/DefaultS3BlockCacheTest.java
index 0c6d300f1a..3a135ac8cc 100644
--- a/core/src/test/java/kafka/log/s3/DefaultS3BlockCacheTest.java
+++ b/core/src/test/java/kafka/log/s3/DefaultS3BlockCacheTest.java
@@ -53,20 +53,22 @@ public void setup() {
     @Test
     public void testRead() throws Exception {
         ObjectWriter objectWriter = new ObjectWriter(0, s3Operator, 1024, 1024);
-        objectWriter.write(newRecord(233, 10, 5, 512));
-        objectWriter.write(newRecord(233, 15, 10, 512));
-        objectWriter.write(newRecord(233, 25, 5, 512));
-        objectWriter.write(newRecord(234, 0, 5, 512));
+        objectWriter.write(233, List.of(
+                newRecord(233, 10, 5, 512),
+                newRecord(233, 15, 10, 512),
+                newRecord(233, 25, 5, 512)
+        ));
+        objectWriter.write(234, List.of(newRecord(234, 0, 5, 512)));
         objectWriter.close();
         S3ObjectMetadata metadata1 = new S3ObjectMetadata(0, objectWriter.size(), S3ObjectType.WAL_LOOSE);
 
         objectWriter = new ObjectWriter(1, s3Operator, 1024, 1024);
-        objectWriter.write(newRecord(233, 30, 10, 512));
+        objectWriter.write(233, List.of(newRecord(233, 30, 10, 512)));
         objectWriter.close();
         S3ObjectMetadata metadata2 = new S3ObjectMetadata(1, objectWriter.size(), S3ObjectType.WAL_LOOSE);
 
         objectWriter = new ObjectWriter(2, s3Operator, 1024, 1024);
-        objectWriter.write(newRecord(233, 40, 20, 512));
+        objectWriter.write(233, List.of(newRecord(233, 40, 20, 512)));
         objectWriter.close();
         S3ObjectMetadata metadata3 = new S3ObjectMetadata(2, objectWriter.size(), S3ObjectType.WAL_LOOSE);
 
diff --git a/core/src/test/java/kafka/log/s3/ObjectWriterTest.java b/core/src/test/java/kafka/log/s3/ObjectWriterTest.java
index 421c5c4609..3816d47b1e 100644
--- a/core/src/test/java/kafka/log/s3/ObjectWriterTest.java
+++ b/core/src/test/java/kafka/log/s3/ObjectWriterTest.java
@@ -43,13 +43,12 @@ public void testWrite() throws ExecutionException, InterruptedException {
         S3Operator s3Operator = new MemoryS3Operator();
         ObjectWriter objectWriter = new ObjectWriter(1, s3Operator, 1024, 1024);
         StreamRecordBatch r1 = newRecord(233, 10, 5, 512);
-        objectWriter.write(r1);
         StreamRecordBatch r2 = newRecord(233, 15, 10, 512);
-        objectWriter.write(r2);
         StreamRecordBatch r3 = newRecord(233, 25, 5, 512);
-        objectWriter.write(r3);
+        objectWriter.write(233, List.of(r1, r2, r3));
+
         StreamRecordBatch r4 = newRecord(234, 0, 5, 512);
-        objectWriter.write(r4);
+        objectWriter.write(234, List.of(r4));
         objectWriter.close().get();
 
         List<ObjectStreamRange> streamRanges = objectWriter.getStreamRanges();
@@ -91,22 +90,20 @@ public void testWrite() throws ExecutionException, InterruptedException {
             assertEquals(25L, r.getBaseOffset());
             assertEquals(5L, r.getRecordBatch().count());
             assertEquals(r3.getRecordBatch().rawPayload(), r.getRecordBatch().rawPayload());
-            r = it.next();
+        }
+
+        blockIndexes = objectReader.find(234, 1, 2).get();
+        assertEquals(1, blockIndexes.size());
+        assertEquals(2, blockIndexes.get(0).blockId());
+        {
+            Iterator<StreamRecordBatch> it = objectReader.read(blockIndexes.get(0)).get().iterator();
+            StreamRecordBatch r = it.next();
             assertEquals(234L, r.getStreamId());
             assertEquals(0L, r.getBaseOffset());
             assertEquals(5L, r.getRecordBatch().count());
             assertEquals(r4.getRecordBatch().rawPayload(), r.getRecordBatch().rawPayload());
             assertFalse(it.hasNext());
         }
-
-        blockIndexes = objectReader.find(234, 1, 2).get();
-        assertEquals(1, blockIndexes.size());
-        assertEquals(1, blockIndexes.get(0).blockId());
-    }
-
-    @Test
-    public void testWrite_closeBlock() {
-
     }
 
     StreamRecordBatch newRecord(long streamId, long offset, int count, int payloadSize) {
diff --git a/core/src/test/java/kafka/log/s3/StreamObjectCopyerTest.java b/core/src/test/java/kafka/log/s3/StreamObjectCopyerTest.java
index 6271610f62..81960541da 100644
--- a/core/src/test/java/kafka/log/s3/StreamObjectCopyerTest.java
+++ b/core/src/test/java/kafka/log/s3/StreamObjectCopyerTest.java
@@ -20,6 +20,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+
 import kafka.log.s3.model.StreamRecordBatch;
 import kafka.log.s3.operator.MemoryS3Operator;
 import kafka.log.s3.operator.S3Operator;
@@ -43,25 +44,23 @@ public void testCompact() throws ExecutionException, InterruptedException {
 
         ObjectWriter objectWriter1 = new ObjectWriter(1, s3Operator, 1024, 1024);
         StreamRecordBatch r1 = newRecord(streamId, 10, 5, 512);
-        objectWriter1.write(r1);
         StreamRecordBatch r2 = newRecord(streamId, 15, 10, 512);
-        objectWriter1.write(r2);
+        objectWriter1.write(streamId, List.of(r1, r2));
         objectWriter1.close().get();
 
         ObjectWriter objectWriter2 = new ObjectWriter(2, s3Operator, 1024, 1024);
         StreamRecordBatch r3 = newRecord(streamId, 25, 8, 512);
-        objectWriter2.write(r3);
         StreamRecordBatch r4 = newRecord(streamId, 33, 6, 512);
-        objectWriter2.write(r4);
+        objectWriter2.write(streamId, List.of(r3, r4));
         objectWriter2.close().get();
 
         S3ObjectMetadata metadata1 = new S3ObjectMetadata(1, objectWriter1.size(), S3ObjectType.STREAM);
         S3ObjectMetadata metadata2 = new S3ObjectMetadata(2, objectWriter2.size(), S3ObjectType.STREAM);
 
         StreamObjectCopyer streamObjectCopyer = new StreamObjectCopyer(targetObjectId,
-            s3Operator,
-            // TODO: use a better clusterName
-            s3Operator.writer(ObjectUtils.genKey(0, "todocluster", targetObjectId))
+                s3Operator,
+                // TODO: use a better clusterName
+                s3Operator.writer(ObjectUtils.genKey(0, "todocluster", targetObjectId))
         );
         streamObjectCopyer.write(metadata1);
         streamObjectCopyer.write(metadata2);