@Test
    public void testUpdateLastEpoch() {
        TopicPartition tp = new TopicPartition("topic-1", 0);

        MetadataResponse metadataResponse = emptyMetadataResponse();
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 0L);

        // if we have no leader epoch, this call shouldn't do anything
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 0));
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 2));
        assertFalse(metadata.lastSeenLeaderEpoch(tp).isPresent());

        // Metadata with newer epoch is handled
        metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 10);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
        assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));

        // Don't update to an older one
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
        assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));

        // Don't cause update if it's the same one
        assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 10));
        assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));

        // Update if we see newer epoch
        assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 12));
        assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));

        metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 12);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 2L);
        assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));

        // Don't overwrite metadata with older epoch
        metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 11);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 3L);
        assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));
    }
Prior to Kafka version 2.4 (which coincides with Metadata version 9), the broker does not propagate leader epoch
information accurately while a reassignment is in progress, so we cannot rely on it. This is explained in more
detail in MetadataResponse's constructor.
clients/src/test/java/org/apache/kafka/clients/MetadataTest.java