private static void processRecoveredRecord(
        StreamRecordBatch streamRecordBatch,
        Map<Long, Long> openingStreamEndOffsets,
        Map<Long, Queue<StreamRecordBatch>> streamDiscontinuousRecords,
        LogCache.LogCacheBlock cacheBlock,
        Map<Long, Long> streamNextOffsets,
        Logger logger
    ) {
        long streamId = streamRecordBatch.getStreamId();

        Long openingStreamEndOffset = openingStreamEndOffsets.get(streamId);
        if (openingStreamEndOffset == null || openingStreamEndOffset > streamRecordBatch.getBaseOffset()) {
            // stream is already safe closed, or the record have been committed, skip it
            streamRecordBatch.release();
            return;
        }

        Long expectedNextOffset = streamNextOffsets.get(streamId);
        Queue<StreamRecordBatch> discontinuousRecords = streamDiscontinuousRecords.get(streamId);
        boolean isContinuous = expectedNextOffset == null || expectedNextOffset == streamRecordBatch.getBaseOffset();
        if (!isContinuous) {
            // unexpected record, put it into discontinuous records queue.
            if (discontinuousRecords == null) {
                discontinuousRecords = new PriorityQueue<>(Comparator.comparingLong(StreamRecordBatch::getBaseOffset));
                streamDiscontinuousRecords.put(streamId, discontinuousRecords);
            }
            discontinuousRecords.add(streamRecordBatch);
            return;
        }
        // continuous record, put it into cache, and check if there is any historical discontinuous records can be polled.
        cacheBlock.put(streamRecordBatch);
        expectedNextOffset = maybePollDiscontinuousRecords(streamRecordBatch, cacheBlock, discontinuousRecords, logger);
        streamNextOffsets.put(streamId, 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