private static long maybePollDiscontinuousRecords(
        StreamRecordBatch streamRecordBatch,
        LogCache.LogCacheBlock cacheBlock,
        Queue<StreamRecordBatch> discontinuousRecords,
        Logger logger
    ) {
        long expectedNextOffset = streamRecordBatch.getLastOffset();
        if (discontinuousRecords == null) {
            return expectedNextOffset;
        }
        // check and poll historical discontinuous records.
        while (!discontinuousRecords.isEmpty()) {
            StreamRecordBatch peek = discontinuousRecords.peek();
            if (peek.getBaseOffset() != expectedNextOffset) {
                break;
            }
            // should never happen, log it.
            logger.error("[BUG] recover an out of order record, streamId={}, expectedNextOffset={}, record={}", streamRecordBatch.getStreamId(), expectedNextOffset, peek);
            discontinuousRecords.poll();
            cacheBlock.put(peek);
            expectedNextOffset = peek.getLastOffset();
        }
        return expectedNextOffset;
    }
Processes recovered stream records. Caches continuous ones or queues discontinuous based on offset order.

@param streamRecordBatch          the recovered record batch to process
@param openingStreamEndOffsets    the end offsets of each opening stream
@param streamDiscontinuousRecords the out-of-order records of each stream (to be filled)
@param cacheBlock                 the cache block (to be filled)
@param streamNextOffsets          the next offsets of each stream (to be updated)
@param logger                     logger
s3stream/src/main/java/com/automq/stream/s3/S3Storage.java