@Test
    public void testEpochUpdateOnChangedTopicIds() {
        TopicPartition tp = new TopicPartition("topic-1", 0);
        Map<String, Uuid> topicIds = Collections.singletonMap("topic-1", Uuid.randomUuid());

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

        // Start with a topic with no topic ID
        metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 100);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
        assertEquals(Optional.of(100), metadata.lastSeenLeaderEpoch(tp));

        // If the older topic ID is null, we should go with the new topic ID as the leader epoch
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 10, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 2L);
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));

        // Don't cause update if it's the same one
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 10, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 3L);
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));

        // Update if we see newer epoch
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 12, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 4L);
        assertEquals(Optional.of(12), metadata.lastSeenLeaderEpoch(tp));

        // We should also update if we see a new topicId even if the epoch is lower
        Map<String, Uuid> newTopicIds = Collections.singletonMap("topic-1", Uuid.randomUuid());
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 3, newTopicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 5L);
        assertEquals(Optional.of(3), metadata.lastSeenLeaderEpoch(tp));

        // Finally, update when the topic ID is new and the epoch is higher
        Map<String, Uuid> newTopicIds2 = Collections.singletonMap("topic-1", Uuid.randomUuid());
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 20, newTopicIds2);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 6L);
        assertEquals(Optional.of(20), metadata.lastSeenLeaderEpoch(tp));
    }
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