@Test
    public void testDeleteRetentionMsOnExpiredSegment() throws RemoteStorageException, IOException {
        AtomicLong logStartOffset = new AtomicLong(0);
        try (RemoteLogManager remoteLogManager = new RemoteLogManager(config.remoteLogManagerConfig(), brokerId, logDir, clusterId, time,
                tp -> Optional.of(mockLog),
                (topicPartition, offset) -> logStartOffset.set(offset),
                brokerTopicStats, metrics) {
            public RemoteStorageManager createRemoteStorageManager() {
                return remoteStorageManager;
            }
            public RemoteLogMetadataManager createRemoteLogMetadataManager() {
                return remoteLogMetadataManager;
            }
        }) {
            RemoteLogManager.RLMExpirationTask task = remoteLogManager.new RLMExpirationTask(leaderTopicIdPartition);

            when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition());
            when(mockLog.logEndOffset()).thenReturn(200L);

            List<EpochEntry> epochEntries = Collections.singletonList(epochEntry0);

            List<RemoteLogSegmentMetadata> remoteLogSegmentMetadatas = listRemoteLogSegmentMetadata(
                    leaderTopicIdPartition, 2, 100, 1024, epochEntries, RemoteLogSegmentState.DELETE_SEGMENT_FINISHED);

            when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition))
                    .thenReturn(remoteLogSegmentMetadatas.iterator());
            when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, 0))
                    .thenReturn(remoteLogSegmentMetadatas.iterator())
                    .thenReturn(remoteLogSegmentMetadatas.iterator());

            checkpoint.write(epochEntries);
            LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint, scheduler);
            when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));

            Map<String, Long> logProps = new HashMap<>();
            logProps.put("retention.bytes", -1L);
            logProps.put("retention.ms", 0L);
            LogConfig mockLogConfig = new LogConfig(logProps);
            when(mockLog.config()).thenReturn(mockLogConfig);

            when(remoteLogMetadataManager.updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class)))
                    .thenAnswer(answer -> CompletableFuture.runAsync(() -> { }));

            task.cleanupExpiredRemoteLogSegments();

            verifyNoMoreInteractions(remoteStorageManager);
            assertEquals(0L, logStartOffset.get());
        } catch (ExecutionException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
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