@Override
    @WithSpan
    public CompletableFuture<Void> append(AppendContext context, StreamRecordBatch streamRecord) {
        final long startTime = System.nanoTime();
        CompletableFuture<Void> cf = new CompletableFuture<>();
        // encoded before append to free heap ByteBuf.
        streamRecord.encoded();
        WalWriteRequest writeRequest = new WalWriteRequest(streamRecord, -1L, cf, context);
        handleAppendRequest(writeRequest);
        append0(context, writeRequest, false);
        return cf.whenComplete((nil, ex) -> {
            streamRecord.release();
            StorageOperationStats.getInstance().appendStats.record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS));
        });
    }
Recover WAL, upload WAL to S3 and close opening streams.
Note: {@link WriteAheadLog#start()} should be called before this method.
s3stream/src/main/java/com/automq/stream/s3/S3Storage.java