public FetchDataInfo read(RemoteStorageFetchInfo remoteStorageFetchInfo) throws RemoteStorageException, IOException {
        int fetchMaxBytes = remoteStorageFetchInfo.fetchMaxBytes;
        TopicPartition tp = remoteStorageFetchInfo.topicPartition;
        FetchRequest.PartitionData fetchInfo = remoteStorageFetchInfo.fetchInfo;

        boolean includeAbortedTxns = remoteStorageFetchInfo.fetchIsolation == FetchIsolation.TXN_COMMITTED;

        long offset = fetchInfo.fetchOffset;
        int maxBytes = Math.min(fetchMaxBytes, fetchInfo.maxBytes);

        Optional<UnifiedLog> logOptional = fetchLog.apply(tp);
        OptionalInt epoch = OptionalInt.empty();

        if (logOptional.isPresent()) {
            Option<LeaderEpochFileCache> leaderEpochCache = logOptional.get().leaderEpochCache();
            if (leaderEpochCache != null && leaderEpochCache.isDefined()) {
                epoch = leaderEpochCache.get().epochForOffset(offset);
            }
        }

        Optional<RemoteLogSegmentMetadata> rlsMetadataOptional = epoch.isPresent()
                ? fetchRemoteLogSegmentMetadata(tp, epoch.getAsInt(), offset)
                : Optional.empty();

        if (!rlsMetadataOptional.isPresent()) {
            String epochStr = (epoch.isPresent()) ? Integer.toString(epoch.getAsInt()) : "NOT AVAILABLE";
            throw new OffsetOutOfRangeException("Received request for offset " + offset + " for leader epoch "
                    + epochStr + " and partition " + tp + " which does not exist in remote tier.");
        }

        RemoteLogSegmentMetadata remoteLogSegmentMetadata = rlsMetadataOptional.get();
        InputStream remoteSegInputStream = null;
        try {
            int startPos = 0;
            RecordBatch firstBatch = null;

            //  Iteration over multiple RemoteSegmentMetadata is required in case of log compaction.
            //  It may be possible the offset is log compacted in the current RemoteLogSegmentMetadata
            //  And we need to iterate over the next segment metadata to fetch messages higher than the given offset.
            while (firstBatch == null && rlsMetadataOptional.isPresent()) {
                remoteLogSegmentMetadata = rlsMetadataOptional.get();
                // Search forward for the position of the last offset that is greater than or equal to the target offset
                startPos = lookupPositionForOffset(remoteLogSegmentMetadata, offset);
                remoteSegInputStream = remoteLogStorageManager.fetchLogSegment(remoteLogSegmentMetadata, startPos);
                RemoteLogInputStream remoteLogInputStream = getRemoteLogInputStream(remoteSegInputStream);
                firstBatch = findFirstBatch(remoteLogInputStream, offset);
                if (firstBatch == null) {
                    rlsMetadataOptional = findNextSegmentMetadata(rlsMetadataOptional.get(), logOptional.get().leaderEpochCache());
                }
            }
            if (firstBatch == null)
                return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY, false,
                        includeAbortedTxns ? Optional.of(Collections.emptyList()) : Optional.empty());

            int firstBatchSize = firstBatch.sizeInBytes();
            // An empty record is sent instead of an incomplete batch when
            //  - there is no minimum-one-message constraint and
            //  - the first batch size is more than maximum bytes that can be sent and
            //  - for FetchRequest version 3 or above.
            if (!remoteStorageFetchInfo.minOneMessage &&
                    !remoteStorageFetchInfo.hardMaxBytesLimit &&
                    firstBatchSize > maxBytes) {
                return new FetchDataInfo(new LogOffsetMetadata(offset), MemoryRecords.EMPTY);
            }

            int updatedFetchSize =
                    remoteStorageFetchInfo.minOneMessage && firstBatchSize > maxBytes ? firstBatchSize : maxBytes;

            ByteBuffer buffer = ByteBuffer.allocate(updatedFetchSize);
            int remainingBytes = updatedFetchSize;

            firstBatch.writeTo(buffer);
            remainingBytes -= firstBatchSize;

            if (remainingBytes > 0) {
                // read the input stream until min of (EOF stream or buffer's remaining capacity).
                Utils.readFully(remoteSegInputStream, buffer);
            }
            buffer.flip();

            FetchDataInfo fetchDataInfo = new FetchDataInfo(
                    new LogOffsetMetadata(offset, remoteLogSegmentMetadata.startOffset(), startPos),
                    MemoryRecords.readableRecords(buffer));
            if (includeAbortedTxns) {
                fetchDataInfo = addAbortedTransactions(firstBatch.baseOffset(), remoteLogSegmentMetadata, fetchDataInfo, logOptional.get());
            }

            return fetchDataInfo;
        } finally {
            Utils.closeQuietly(remoteSegInputStream, "RemoteLogSegmentInputStream");
        }
    }
Returns a map containing the epoch vs start-offset for the given leader epoch map by filtering the epochs that
does not contain any messages/records associated with them.
For ex:
<pre>
{@code
 <epoch - start offset>
 0 - 0
 1 - 10
 2 - 20
 3 - 30
 4 - 40
 5 - 60  // epoch 5 does not have records or messages associated with it
 6 - 60
 7 - 70
}
</pre>
When the above leaderEpochMap is passed to this method, it returns the following map:
<pre>
{@code
 <epoch - start offset>
 0 - 0
 1 - 10
 2 - 20
 3 - 30
 4 - 40
 6 - 60
 7 - 70
}
</pre>
@param leaderEpochs The leader epoch map to be refined.
core/src/main/java/kafka/log/remote/RemoteLogManager.java