@Test
    public void testCopyThrottling() throws Exception {
        long oldestSegmentStartOffset = 0L;

        when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition());

        // leader epoch preparation
        checkpoint.write(totalEpochEntries);
        LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint, scheduler);
        when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));
        when(remoteLogMetadataManager.highestOffsetForEpoch(any(TopicIdPartition.class), anyInt())).thenReturn(Optional.of(0L));

        // create 3 log segments
        LogSegment segmentToCopy = mock(LogSegment.class);
        LogSegment segmentToThrottle = mock(LogSegment.class);
        LogSegment activeSegment = mock(LogSegment.class);

        File tempFile = TestUtils.tempFile();
        FileRecords fileRecords = mock(FileRecords.class);
        when(fileRecords.file()).thenReturn(tempFile);
        when(fileRecords.sizeInBytes()).thenReturn(10);

        // set up the segment that will be copied
        when(segmentToCopy.log()).thenReturn(fileRecords);
        when(segmentToCopy.baseOffset()).thenReturn(oldestSegmentStartOffset);
        when(segmentToCopy.readNextOffset()).thenReturn(100L);

        // set up the segment that will not be copied because of hitting quota
        when(segmentToThrottle.log()).thenReturn(fileRecords);
        when(segmentToThrottle.baseOffset()).thenReturn(100L);
        when(segmentToThrottle.readNextOffset()).thenReturn(150L);

        // set up the active segment
        when(activeSegment.log()).thenReturn(fileRecords);
        when(activeSegment.baseOffset()).thenReturn(150L);

        when(mockLog.activeSegment()).thenReturn(activeSegment);
        when(mockLog.logStartOffset()).thenReturn(oldestSegmentStartOffset);
        when(mockLog.logSegments(anyLong(), anyLong())).thenReturn(JavaConverters.collectionAsScalaIterable(Arrays.asList(segmentToCopy, segmentToThrottle, activeSegment)));

        File mockProducerSnapshotIndex = TestUtils.tempFile();
        ProducerStateManager mockStateManager = mock(ProducerStateManager.class);
        when(mockStateManager.fetchSnapshot(anyLong())).thenReturn(Optional.of(mockProducerSnapshotIndex));

        when(mockLog.producerStateManager()).thenReturn(mockStateManager);
        when(mockLog.lastStableOffset()).thenReturn(250L);

        File tempDir = TestUtils.tempDirectory();
        OffsetIndex idx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, oldestSegmentStartOffset, ""), oldestSegmentStartOffset, 1000).get();
        TimeIndex timeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, oldestSegmentStartOffset, ""), oldestSegmentStartOffset, 1500).get();
        File txnFile = UnifiedLog.transactionIndexFile(tempDir, oldestSegmentStartOffset, "");
        txnFile.createNewFile();
        TransactionIndex txnIndex = new TransactionIndex(oldestSegmentStartOffset, txnFile);
        when(segmentToCopy.timeIndex()).thenReturn(timeIdx);
        when(segmentToCopy.offsetIndex()).thenReturn(idx);
        when(segmentToCopy.txnIndex()).thenReturn(txnIndex);

        CompletableFuture<Void> dummyFuture = new CompletableFuture<>();
        dummyFuture.complete(null);
        when(remoteLogMetadataManager.addRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadata.class))).thenReturn(dummyFuture);
        when(remoteLogMetadataManager.updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class))).thenReturn(dummyFuture);
        when(remoteStorageManager.copyLogSegmentData(any(RemoteLogSegmentMetadata.class), any(LogSegmentData.class))).thenReturn(Optional.empty());

        // After the first call, getThrottleTimeMs should return non-zero throttle time
        when(rlmCopyQuotaManager.getThrottleTimeMs()).thenReturn(0L, 1000L);
        doNothing().when(rlmCopyQuotaManager).record(anyInt());

        RemoteLogManager.RLMCopyTask task = remoteLogManager.new RLMCopyTask(leaderTopicIdPartition, 128);

        // Verify that the copy operation times out, since the second segment cannot be copied due to quota being exceeded
        assertThrows(AssertionFailedError.class, () -> assertTimeoutPreemptively(Duration.ofMillis(200), () -> task.copyLogSegmentsToRemote(mockLog)));

        // Verify the highest offset in remote storage is updated corresponding to the only segment that was copied
        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 99L after the copy
        assertEquals(99L, 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