@ParameterizedTest
    @ValueSource(booleans = {false, true})
    public void testCopyQuota(boolean quotaExceeded) throws Exception {
        RemoteLogManager.RLMCopyTask task = setupRLMTask(quotaExceeded);

        if (quotaExceeded) {
            // Verify that the copy operation times out, since no segments can be copied due to quota being exceeded
            assertThrows(AssertionFailedError.class, () -> assertTimeoutPreemptively(Duration.ofMillis(200), () -> task.copyLogSegmentsToRemote(mockLog)));

            Map<org.apache.kafka.common.MetricName, KafkaMetric> allMetrics = metrics.metrics();
            KafkaMetric avgMetric = allMetrics.get(metrics.metricName("remote-copy-throttle-time-avg", "RemoteLogManager"));
            KafkaMetric maxMetric = allMetrics.get(metrics.metricName("remote-copy-throttle-time-max", "RemoteLogManager"));
            assertEquals(quotaExceededThrottleTime, ((Double) avgMetric.metricValue()).longValue());
            assertEquals(quotaExceededThrottleTime, ((Double) maxMetric.metricValue()).longValue());

            // Verify the highest offset in remote storage is updated only once
            ArgumentCaptor<Long> capture = ArgumentCaptor.forClass(Long.class);
            verify(mockLog, times(1)).updateHighestOffsetInRemoteStorage(capture.capture());
            // Verify the highest offset in remote storage was -1L before the copy started
            assertEquals(-1L, capture.getValue());
        } else {
            // Verify the copy operation completes within the timeout, since it does not need to wait for quota availability
            assertTimeoutPreemptively(Duration.ofMillis(100), () -> task.copyLogSegmentsToRemote(mockLog));

            // Verify quota check was performed
            verify(rlmCopyQuotaManager, times(1)).getThrottleTimeMs();
            // Verify bytes to copy was recorded with the quota manager
            verify(rlmCopyQuotaManager, times(1)).record(10);

            Map<org.apache.kafka.common.MetricName, KafkaMetric> allMetrics = metrics.metrics();
            KafkaMetric avgMetric = allMetrics.get(metrics.metricName("remote-copy-throttle-time-avg", "RemoteLogManager"));
            KafkaMetric maxMetric = allMetrics.get(metrics.metricName("remote-copy-throttle-time-max", "RemoteLogManager"));
            assertEquals(Double.NaN, avgMetric.metricValue());
            assertEquals(Double.NaN, maxMetric.metricValue());

            // Verify the highest offset in remote storage is updated
            ArgumentCaptor<Long> capture = ArgumentCaptor.forClass(Long.class);
            verify(mockLog, times(2)).updateHighestOffsetInRemoteStorage(capture.capture());
            List<Long> capturedValues = capture.getAllValues();
            // Verify the highest offset in remote storage was -1L before the copy
            assertEquals(-1L, capturedValues.get(0).longValue());
            // Verify it was updated to 149L after the copy
            assertEquals(149L, capturedValues.get(1).longValue());
        }
    }
This test asserts that the newly elected leader for a partition is able to find the log-start-offset.
Note that the case tested here is that the previous leader deleted the log segments up-to offset 500. And, the
log-start-offset didn't propagate to the replicas before the leader-election.
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java