@Test
    void testRemoteLogManagerRemoteMetrics() throws Exception {
        remoteLogManager.startup();
        long oldestSegmentStartOffset = 0L;
        long olderSegmentStartOffset = 75L;
        long nextSegmentStartOffset = 150L;
        when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition());
        when(mockLog.parentDir()).thenReturn("dir1");

        // 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));

        File tempFile = TestUtils.tempFile();
        File mockProducerSnapshotIndex = TestUtils.tempFile();
        File tempDir = TestUtils.tempDirectory();
        // create 3 log segments, with 0, 75 and 150 as log start offset
        LogSegment oldestSegment = mock(LogSegment.class);
        LogSegment olderSegment = mock(LogSegment.class);
        LogSegment activeSegment = mock(LogSegment.class);

        when(oldestSegment.baseOffset()).thenReturn(oldestSegmentStartOffset);
        when(olderSegment.baseOffset()).thenReturn(olderSegmentStartOffset);
        when(activeSegment.baseOffset()).thenReturn(nextSegmentStartOffset);

        FileRecords oldestFileRecords = mock(FileRecords.class);
        when(oldestSegment.log()).thenReturn(oldestFileRecords);
        when(oldestFileRecords.file()).thenReturn(tempFile);
        when(oldestFileRecords.sizeInBytes()).thenReturn(10);
        when(oldestSegment.readNextOffset()).thenReturn(olderSegmentStartOffset);

        FileRecords olderFileRecords = mock(FileRecords.class);
        when(olderSegment.log()).thenReturn(olderFileRecords);
        when(olderFileRecords.file()).thenReturn(tempFile);
        when(olderFileRecords.sizeInBytes()).thenReturn(10);
        when(olderSegment.readNextOffset()).thenReturn(nextSegmentStartOffset);

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

        ProducerStateManager mockStateManager = mock(ProducerStateManager.class);
        when(mockLog.producerStateManager()).thenReturn(mockStateManager);
        when(mockStateManager.fetchSnapshot(anyLong())).thenReturn(Optional.of(mockProducerSnapshotIndex));
        when(mockLog.lastStableOffset()).thenReturn(250L);
        Map<String, Long> logProps = new HashMap<>();
        logProps.put("retention.bytes", 1000000L);
        logProps.put("retention.ms", -1L);
        LogConfig logConfig = new LogConfig(logProps);
        when(mockLog.config()).thenReturn(logConfig);

        OffsetIndex oldestIdx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, oldestSegmentStartOffset, ""), oldestSegmentStartOffset, 1000).get();
        TimeIndex oldestTimeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, oldestSegmentStartOffset, ""), oldestSegmentStartOffset, 1500).get();
        File oldestTxnFile = UnifiedLog.transactionIndexFile(tempDir, oldestSegmentStartOffset, "");
        oldestTxnFile.createNewFile();
        TransactionIndex oldestTxnIndex = new TransactionIndex(oldestSegmentStartOffset, oldestTxnFile);
        when(oldestSegment.timeIndex()).thenReturn(oldestTimeIdx);
        when(oldestSegment.offsetIndex()).thenReturn(oldestIdx);
        when(oldestSegment.txnIndex()).thenReturn(oldestTxnIndex);

        OffsetIndex olderIdx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, olderSegmentStartOffset, ""), olderSegmentStartOffset, 1000).get();
        TimeIndex olderTimeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, olderSegmentStartOffset, ""), olderSegmentStartOffset, 1500).get();
        File olderTxnFile = UnifiedLog.transactionIndexFile(tempDir, olderSegmentStartOffset, "");
        oldestTxnFile.createNewFile();
        TransactionIndex olderTxnIndex = new TransactionIndex(olderSegmentStartOffset, olderTxnFile);
        when(olderSegment.timeIndex()).thenReturn(olderTimeIdx);
        when(olderSegment.offsetIndex()).thenReturn(olderIdx);
        when(olderSegment.txnIndex()).thenReturn(olderTxnIndex);

        CompletableFuture<Void> dummyFuture = new CompletableFuture<>();
        dummyFuture.complete(null);
        when(remoteLogMetadataManager.addRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadata.class))).thenReturn(dummyFuture);
        when(remoteLogMetadataManager.updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class))).thenReturn(dummyFuture);
        Iterator<RemoteLogSegmentMetadata> iterator = listRemoteLogSegmentMetadata(leaderTopicIdPartition, 5, 100, 1024, RemoteLogSegmentState.COPY_SEGMENT_FINISHED).iterator();
        when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition)).thenReturn(iterator);
        when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, 2)).thenReturn(iterator);
        when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, 1)).thenReturn(iterator);

        CountDownLatch remoteLogSizeComputationTimeLatch = new CountDownLatch(1);
        when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, 0)).thenAnswer(ans -> {
            // advance the mock timer 1000ms to add value for RemoteLogSizeComputationTime metric
            time.sleep(1000);
            return iterator;
        }).thenAnswer(ans -> {
            // wait for verifying RemoteLogSizeComputationTime metric value.
            remoteLogSizeComputationTimeLatch.await(5000, TimeUnit.MILLISECONDS);
            return Collections.emptyIterator();
        });

        CountDownLatch latch = new CountDownLatch(1);
        doAnswer(ans -> Optional.empty()).doAnswer(ans -> {
            // waiting for verification
            latch.await(5000, TimeUnit.MILLISECONDS);
            return Optional.empty();
        }).when(remoteStorageManager).copyLogSegmentData(any(RemoteLogSegmentMetadata.class), any(LogSegmentData.class));
        when(rlmCopyQuotaManager.getThrottleTimeMs()).thenReturn(quotaAvailableThrottleTime);

        Partition mockLeaderPartition = mockPartition(leaderTopicIdPartition);

        // This method is called by both Copy and Expiration task. On the first call, both tasks should see 175 bytes as
        // the local log segments size
        when(mockLog.onlyLocalLogSegmentsSize()).thenReturn(175L, 175L, 100L);
        when(activeSegment.size()).thenReturn(100);
        when(mockLog.onlyLocalLogSegmentsCount()).thenReturn(2L).thenReturn(1L);

        // before running tasks, the metric should not be registered
        assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic));
        assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic));
        assertThrows(NoSuchElementException.class, () -> yammerMetricValue("RemoteLogSizeComputationTime,topic=" + leaderTopic));
        // all topic metrics should be 0
        assertEquals(0L, yammerMetricValue("RemoteCopyLagBytes"));
        assertEquals(0L, yammerMetricValue("RemoteCopyLagSegments"));
        assertEquals(0L, yammerMetricValue("RemoteLogSizeComputationTime"));

        remoteLogManager.onLeadershipChange(Collections.singleton(mockLeaderPartition), Collections.emptySet(), topicIds);
        TestUtils.waitForCondition(
                () -> 75 == safeLongYammerMetricValue("RemoteCopyLagBytes") && 75 == safeLongYammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic),
                String.format("Expected to find 75 for RemoteCopyLagBytes metric value, but found %d for topic 'Leader' and %d for all topics.",
                        safeLongYammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic),
                        safeLongYammerMetricValue("RemoteCopyLagBytes")));
        TestUtils.waitForCondition(
                () -> 1 == safeLongYammerMetricValue("RemoteCopyLagSegments") && 1 == safeLongYammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic),
                String.format("Expected to find 1 for RemoteCopyLagSegments metric value, but found %d for topic 'Leader' and %d for all topics.",
                        safeLongYammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic),
                        safeLongYammerMetricValue("RemoteCopyLagSegments")));
        // unlock copyLogSegmentData
        latch.countDown();

        TestUtils.waitForCondition(
                () -> safeLongYammerMetricValue("RemoteLogSizeComputationTime") >= 1000 && safeLongYammerMetricValue("RemoteLogSizeComputationTime,topic=" + leaderTopic) >= 1000,
                String.format("Expected to find 1000 for RemoteLogSizeComputationTime metric value, but found %d for topic 'Leader' and %d for all topics.",
                        safeLongYammerMetricValue("RemoteLogSizeComputationTime,topic=" + leaderTopic),
                        safeLongYammerMetricValue("RemoteLogSizeComputationTime")));
        remoteLogSizeComputationTimeLatch.countDown();
        
        TestUtils.waitForCondition(
                () -> 0 == safeLongYammerMetricValue("RemoteCopyLagBytes") && 0 == safeLongYammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic),
                String.format("Expected to find 0 for RemoteCopyLagBytes metric value, but found %d for topic 'Leader' and %d for all topics.",
                        safeLongYammerMetricValue("RemoteCopyLagBytes,topic=" + leaderTopic),
                        safeLongYammerMetricValue("RemoteCopyLagBytes")));
        TestUtils.waitForCondition(
                () -> 0 == safeLongYammerMetricValue("RemoteCopyLagSegments") && 0 == safeLongYammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic),
                String.format("Expected to find 0 for RemoteCopyLagSegments metric value, but found %d for topic 'Leader' and %d for all topics.",
                        safeLongYammerMetricValue("RemoteCopyLagSegments,topic=" + leaderTopic),
                        safeLongYammerMetricValue("RemoteCopyLagSegments")));
    }
The following values will be equal when the active segment gets rotated to passive and there are no new messages:
last-stable-offset = high-water-mark = log-end-offset = base-offset-of-active-segment.
This test asserts that the active log segment that was rotated after log.roll.ms are copied to remote storage.
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java