private static void releaseDiscontinuousRecords(Map<Long, Queue<StreamRecordBatch>> streamDiscontinuousRecords, Logger logger) {
        streamDiscontinuousRecords.values().stream()
            .filter(q -> !q.isEmpty())
            .peek(q -> logger.info("drop discontinuous records, records={}", q))
            .forEach(S3Storage::releaseRecords);
    }
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