public List<WalWriteRequest> after(WalWriteRequest request) {
            request.persisted = true;

            // Try to pop sequential persisted requests from the queue.
            long streamId = request.record.getStreamId();
            Queue<WalWriteRequest> streamRequests = stream2requests.get(streamId);
            WalWriteRequest peek = streamRequests.peek();
            if (peek == null || peek.offset != request.offset) {
                return Collections.emptyList();
            }

            LinkedList<WalWriteRequest> rst = new LinkedList<>();
            WalWriteRequest poll = streamRequests.poll();
            assert poll == peek;
            rst.add(poll);

            for (; ; ) {
                peek = streamRequests.peek();
                if (peek == null || !peek.persisted) {
                    break;
                }
                poll = streamRequests.poll();
                assert poll == peek;
                assert poll.record.getBaseOffset() == rst.getLast().record.getLastOffset();
                rst.add(poll);
            }

            return rst;
        }
Try pop sequence persisted request from stream queue and move forward wal inclusive confirm offset.
When the {@code request.record.getStreamId()} is different, concurrent calls are allowed.
When the {@code request.record.getStreamId()} is the same, concurrent calls are not allowed.

@return popped sequence persisted request.
s3stream/src/main/java/com/automq/stream/s3/S3Storage.java