@Override
        public RecordBatch nextBatch() throws IOException {
            for (; ; ) {
                RecordBatch recordBatch = remaining.poll();
                if (recordBatch != null) {
                    return recordBatch;
                }
                if (readSize > maxSize || nextFetchOffset >= endOffset) {
                    return null;
                }
                try {
                    FetchResult rst = elasticLogFileRecords.streamSlice.fetch(fetchContext, nextFetchOffset, endOffset, Math.min(maxSize - readSize, FETCH_BATCH_SIZE)).get();
                    rst.recordBatchList().forEach(streamRecord -> {
                        try {
                            ByteBuffer buf = streamRecord.rawPayload();
                            if (buf.isDirect()) {
                                ByteBuffer heapBuf = ByteBuffer.allocate(buf.remaining());
                                heapBuf.put(buf);
                                heapBuf.flip();
                                buf = heapBuf;
                            }
                            readSize += buf.remaining();
                            for (RecordBatch r : MemoryRecords.readableRecords(buf).batches()) {
                                remaining.offer(r);
                                nextFetchOffset = r.lastOffset() - elasticLogFileRecords.baseOffset + 1;
                            }
                        } catch (Throwable e) {
                            ElasticStreamSlice slice = elasticLogFileRecords.streamSlice;
                            byte[] bytes = new byte[streamRecord.rawPayload().remaining()];
                            streamRecord.rawPayload().get(bytes);
                            LOGGER.error("next batch parse error, stream={} baseOffset={} payload={}", slice.stream().streamId(), slice.sliceRange().start() + streamRecord.baseOffset(), bytes);
                            throw new RuntimeException(e);
                        }
                    });
                    rst.free();
                    if (remaining.isEmpty()) {
                        return null;
                    }
                } catch (Throwable e) {
                    throw new IOException(e);
                }
            }
        }
Return the largest timestamp of the messages after a given offset

@param startOffset The starting offset.
@return The largest timestamp of the messages after the given position.
core/src/main/scala/kafka/log/streamaspect/ElasticLogFileRecords.java