private static RecoveryBlockResult filterOutInvalidStreams(LogCache.LogCacheBlock cacheBlock,
        Map<Long, Long> openingStreamEndOffsets) {
        Set<Long> invalidStreams = new HashSet<>();
        List<RuntimeException> exceptions = new ArrayList<>();

        cacheBlock.records().forEach((streamId, records) -> {
            if (!records.isEmpty()) {
                long startOffset = records.get(0).getBaseOffset();
                long expectedStartOffset = openingStreamEndOffsets.getOrDefault(streamId, startOffset);
                if (startOffset != expectedStartOffset) {
                    RuntimeException exception = new IllegalStateException(String.format("[BUG] WAL data may lost, streamId %d endOffset=%d from controller, " +
                        "but WAL recovered records startOffset=%s", streamId, expectedStartOffset, startOffset));
                    LOGGER.error("invalid stream records", exception);
                    invalidStreams.add(streamId);
                    exceptions.add(exception);
                }
            }
        });

        if (invalidStreams.isEmpty()) {
            return new RecoveryBlockResult(cacheBlock, null);
        }

        // Only streams not in invalidStreams should be uploaded and closed,
        // so re-new a cache block and put only valid records into it, and release all invalid records.
        LogCache.LogCacheBlock newCacheBlock = new LogCache.LogCacheBlock(1024L * 1024 * 1024);
        cacheBlock.records().forEach((streamId, records) -> {
            if (!invalidStreams.contains(streamId)) {
                records.forEach(newCacheBlock::put);
            } else {
                // release invalid records.
                releaseRecords(records);
            }
        });
        return new RecoveryBlockResult(newCacheBlock, ExceptionUtil.combine(exceptions));
    }
Filter out invalid streams (the recovered start offset mismatches the stream end offset from controller) from the cache block if there are any.
s3stream/src/main/java/com/automq/stream/s3/S3Storage.java