@Test
    public void testTierLagResetsToZeroOnBecomingFollower() {
        remoteLogManager.startup();
        remoteLogManager.onLeadershipChange(
                Collections.singleton(mockPartition(leaderTopicIdPartition)), Collections.emptySet(), topicIds);
        RemoteLogManager.RLMCopyTask rlmTask = (RemoteLogManager.RLMCopyTask) remoteLogManager.rlmCopyTask(leaderTopicIdPartition);
        assertNotNull(rlmTask);
        rlmTask.recordLagStats(1024, 2);
        assertEquals(1024, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagBytes());
        assertEquals(2, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());
        // The same node becomes follower now which was the previous leader
        remoteLogManager.onLeadershipChange(Collections.emptySet(),
                Collections.singleton(mockPartition(leaderTopicIdPartition)), topicIds);
        assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagBytes());
        assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());

        // If the old task emits the tier-lag stats, then it should be discarded
        rlmTask.recordLagStats(2048, 4);
        assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagBytes());
        assertEquals(0, brokerTopicStats.topicStats(leaderTopicIdPartition.topic()).remoteCopyLagSegments());
    }
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