@Override
    public Void call() {
        RemoteLogReadResult result;
        try {
            LOGGER.debug("Reading records from remote storage for topic partition {}", fetchInfo.topicPartition);
            FetchDataInfo fetchDataInfo = remoteReadTimer.time(() -> rlm.read(fetchInfo));
            brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).remoteFetchBytesRate().mark(fetchDataInfo.records.sizeInBytes());
            brokerTopicStats.allTopicsStats().remoteFetchBytesRate().mark(fetchDataInfo.records.sizeInBytes());
            result = new RemoteLogReadResult(Optional.of(fetchDataInfo), Optional.empty());
        } catch (OffsetOutOfRangeException e) {
            result = new RemoteLogReadResult(Optional.empty(), Optional.of(e));
        } catch (Exception e) {
            brokerTopicStats.topicStats(fetchInfo.topicPartition.topic()).failedRemoteFetchRequestRate().mark();
            brokerTopicStats.allTopicsStats().failedRemoteFetchRequestRate().mark();
            LOGGER.error("Error occurred while reading the remote data for {}", fetchInfo.topicPartition, e);
            result = new RemoteLogReadResult(Optional.empty(), Optional.of(e));
        }
        LOGGER.debug("Finished reading records from remote storage for topic partition {}", fetchInfo.topicPartition);
        quotaManager.record(result.fetchDataInfo.map(fetchDataInfo -> fetchDataInfo.records.sizeInBytes()).orElse(0));
        callback.accept(result);
        return null;
    }
@Test
    public void testRemoteLogReaderWithError() throws RemoteStorageException, IOException {
        when(mockRLM.read(any(RemoteStorageFetchInfo.class))).thenThrow(new RuntimeException("error"));

        Consumer<RemoteLogReadResult> callback = mock(Consumer.class);
        RemoteStorageFetchInfo remoteStorageFetchInfo = new RemoteStorageFetchInfo(0, false, new TopicPartition(TOPIC, 0), null, null, false);
        RemoteLogReader remoteLogReader =
                new RemoteLogReader(remoteStorageFetchInfo, mockRLM, callback, brokerTopicStats, mockQuotaManager, timer);
        remoteLogReader.call();

        // verify the callback did get invoked with the expected remoteLogReadResult
        ArgumentCaptor<RemoteLogReadResult> remoteLogReadResultArg = ArgumentCaptor.forClass(RemoteLogReadResult.class);
        verify(callback, times(1)).accept(remoteLogReadResultArg.capture());
        RemoteLogReadResult actualRemoteLogReadResult = remoteLogReadResultArg.getValue();
        assertTrue(actualRemoteLogReadResult.error.isPresent());
        assertFalse(actualRemoteLogReadResult.fetchDataInfo.isPresent());

        // verify the record method on quota manager was called with the expected value
        ArgumentCaptor<Double> recordedArg = ArgumentCaptor.forClass(Double.class);
        verify(mockQuotaManager, times(1)).record(recordedArg.capture());
        assertEquals(0, recordedArg.getValue());

        // Verify metrics for remote reads are updated correctly
        assertEquals(1, brokerTopicStats.topicStats(TOPIC).remoteFetchRequestRate().count());
        assertEquals(0, brokerTopicStats.topicStats(TOPIC).remoteFetchBytesRate().count());
        assertEquals(1, brokerTopicStats.topicStats(TOPIC).failedRemoteFetchRequestRate().count());
        // Verify aggregate metrics
        assertEquals(1, brokerTopicStats.allTopicsStats().remoteFetchRequestRate().count());
        assertEquals(0, brokerTopicStats.allTopicsStats().remoteFetchBytesRate().count());
        assertEquals(1, brokerTopicStats.allTopicsStats().failedRemoteFetchRequestRate().count());
    }