private CompletableFuture<Void> fetch0(FetchContext context, long startOffset, long endOffset, int maxSize, List<FetchResult> results) {
        if (startOffset >= endOffset || maxSize <= 0) {
            return CompletableFuture.completedFuture(null);
        }
        int adjustedMaxSize = Math.min(maxSize, 1024 * 1024);
        return streamSlice.fetch(context, startOffset, endOffset, adjustedMaxSize)
                .thenCompose(rst -> {
                    results.add(rst);
                    long nextFetchOffset = startOffset;
                    int readSize = 0;
                    for (RecordBatchWithContext recordBatchWithContext : rst.recordBatchList()) {
                        if (recordBatchWithContext.lastOffset() > nextFetchOffset) {
                            nextFetchOffset = recordBatchWithContext.lastOffset();
                        } else {
                            LOGGER.error("Invalid record batch, last offset {} is less than next offset {}",
                                    recordBatchWithContext.lastOffset(), nextFetchOffset);
                            throw new IllegalStateException();
                        }
                        readSize += recordBatchWithContext.rawPayload().remaining();
                    }
                    if (readSize == 0) {
                        return CompletableFuture.completedFuture(null);
                    }
                    return fetch0(context, nextFetchOffset, endOffset, maxSize - readSize, results);
                });
    }
Fetch records from the {@link ElasticStreamSlice}

@param context fetch context
@param startOffset start offset
@param endOffset end offset
@param maxSize max size of the fetched records
@param results result list to be filled
@return a future that completes when reaching the end offset or the max size
core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java