Repository: apache/kafka
File: storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java (Java)

Code before:
    }

    @Test
    public void testRetentionSizeInPercentMetrics() throws RemoteStorageException {
        RemoteLogManager.RLMExpirationTask expirationTask = remoteLogManager.new RLMExpirationTask(leaderTopicIdPartition);

        // Mock remote log segments for size calculation (10 segments * 1024 bytes = 10240 bytes)
        // Use only epochEntry0 to ensure segments are within the leader epoch lineage
        List<EpochEntry> singleEpochEntry = List.of(epochEntry0);
        List<RemoteLogSegmentMetadata> metadataList = listRemoteLogSegmentMetadata(leaderTopicIdPartition, 10,
                100, 1024, singleEpochEntry, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);

        when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition))
                .thenAnswer(ans -> metadataList.iterator());
        when(remoteLogMetadataManager.listRemoteLogSegments(eq(leaderTopicIdPartition), anyInt()))
                .thenAnswer(ans -> metadataList.iterator());

        TreeMap<Integer, Long> epochEntries = new TreeMap<>();
        epochEntries.put(epochEntry0.epoch(), epochEntry0.startOffset());

        // Test case 1: Testing RetentionSizeInPercent metric (standard retention scenario)
        // retentionSize = 12288, onlyLocalLogSegmentsSize = 100, localLogSegmentsSize = 100
        // Each remote log segment size is 1024. There are 10 remote-log-segments. Total remote size = 10 * 1024 = 10240
        // ((100 + 10240) * 100) / 12288 = 84%
        expirationTask.buildRetentionSizeData(12288, 100, 100, 1000, epochEntries, 6144, 12288);
        assertEquals(84, expirationTask.retentionSizeInPercent());

        // Test case 2: Testing LocalRetentionSizeInPercent metric (local retention scenario)
        // localRetentionBytes = 200, localLogSegmentsSize = 100, so percentage = (100 * 100) / 200 = 50%
        expirationTask.buildRetentionSizeData(12288, 100, 100, 1000, epochEntries, 200, 12288);
        assertEquals(50, expirationTask.localRetentionSizeInPercent());

        // Test case 3: Test retentionSizeInPercent metric >= 100%
        // 10 * 1024 (remote) + 3000 = 13240 / 12288 = 107%
        expirationTask.buildRetentionSizeData(12288, 3000, 4000, 1000, epochEntries, 5000, 12288);
        assertEquals(107, expirationTask.retentionSizeInPercent());
        assertFalse(expirationTask.isAllSegmentsValid());

        // Repeat test-case 3 with valid fullCopyFinishedSegmentSizeInBytes
        expirationTask.buildRetentionSizeData(12288, 3000, 4000, 1000, epochEntries, 5000, 10240);
        assertEquals(107, expirationTask.retentionSizeInPercent());
        assertTrue(expirationTask.isAllSegmentsValid());

        // Repeat test-case 3, once all the segments are valid.
        // 10 * 1024 (remote) + 2048 = 12288 / 12288 = 100%
        expirationTask.buildRetentionSizeData(12288, 2048, 3000, 1000, epochEntries, 5000, 10240);
        assertEquals(100, expirationTask.retentionSizeInPercent());
        assertTrue(expirationTask.isAllSegmentsValid());
    }

    @Test

Review comment (bug):
shall we also check the `localRetentionSizeInPercent` for each of the test case?

Better to retrieve the metric value and perform the assertion? You can use `yammerMetricValue` to retrieve the gauge values. 

Code after:
    }

    @Test
    public void testRetentionSizeInPercentMetrics() throws RemoteStorageException {
        RemoteLogManager.RLMExpirationTask expirationTask = remoteLogManager.new RLMExpirationTask(leaderTopicIdPartition);

        // Mock remote log segments for size calculation (10 segments * 1024 bytes = 10240 bytes)
        // Use only epochEntry0 to ensure segments are within the leader epoch lineage
        List<EpochEntry> singleEpochEntry = List.of(epochEntry0);
        List<RemoteLogSegmentMetadata> metadataList = listRemoteLogSegmentMetadata(leaderTopicIdPartition, 10,
                100, 1024, singleEpochEntry, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);

        when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition))
                .thenAnswer(ans -> metadataList.iterator());
        when(remoteLogMetadataManager.listRemoteLogSegments(eq(leaderTopicIdPartition), anyInt()))
                .thenAnswer(ans -> metadataList.iterator());

        TreeMap<Integer, Long> epochEntries = new TreeMap<>();
        epochEntries.put(epochEntry0.epoch(), epochEntry0.startOffset());

        // Register metrics to expose them via JMX
        expirationTask.registerMetrics();

        String retentionMetricName = "name=RetentionSizeInPercent,partition=" + leaderTopicIdPartition.partition() + ",topic=" + leaderTopicIdPartition.topic();
        String localRetentionMetricName = "name=LocalRetentionSizeInPercent,partition=" + leaderTopicIdPartition.partition() + ",topic=" + leaderTopicIdPartition.topic();

        // Test case 1: Testing RetentionSizeInPercent metric (standard retention scenario)
        // retentionSize = 12288, onlyLocalLogSegmentsSize = 100, localLogSegmentsSize = 100
        // Each remote log segment size is 1024. There are 10 remote-log-segments. Total remote size = 10 * 1024 = 10240
        // RetentionSizeInPercent = ((100 + 10240) * 100) / 12288 = 84%
        // LocalRetentionSizeInPercent = (100 * 100) / 6144 = 1%
        expirationTask.buildRetentionSizeData(12288, 100, 100, 1000, epochEntries, 6144, 12288);
        assertEquals(84, yammerMetricValue(retentionMetricName));
        assertEquals(1, yammerMetricValue(localRetentionMetricName));

        // Test case 2: Testing LocalRetentionSizeInPercent metric (local retention scenario)
        // localRetentionBytes = 200, localLogSegmentsSize = 100, so percentage = (100 * 100) / 200 = 50%
        expirationTask.buildRetentionSizeData(12288, 100, 100, 1000, epochEntries, 200, 12288);
        assertEquals(84, yammerMetricValue(retentionMetricName));
        assertEquals(50, yammerMetricValue(localRetentionMetricName));

        // Test case 3: Test retentionSizeInPercent metric >= 100%
        // 10 * 1024 (remote) + 3000 = 13240 / 12288 = 107%
        // LocalRetentionSizeInPercent = (4000 * 100) / 5000 = 80%
        expirationTask.buildRetentionSizeData(12288, 3000, 4000, 1000, epochEntries, 5000, 12288);
        assertEquals(107, yammerMetricValue(retentionMetricName));
        assertEquals(80, yammerMetricValue(localRetentionMetricName));
        assertFalse(expirationTask.isAllSegmentsValid());

        // Repeat test-case 3 with valid fullCopyFinishedSegmentSizeInBytes
        expirationTask.buildRetentionSizeData(12288, 3000, 4000, 1000, epochEntries, 5000, 10240);