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

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

        // Start with a Topic topic-1 with a random topic ID
        Map<String, Uuid> topicIds = Collections.singletonMap("topic-1", Uuid.randomUuid());
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 10, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));

        // Topic topic-1 is now deleted so Response contains an Error. LeaderEpoch should still maintain Old value
        metadataResponse = RequestTestUtils.metadataUpdateWith("dummy", 1, Collections.singletonMap("topic-1", Errors.UNKNOWN_TOPIC_OR_PARTITION), Collections.emptyMap());
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
        assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));

        // Create topic-1 again but this time with a different topic ID. LeaderEpoch should be updated to new even if lower.
        Map<String, Uuid> newTopicIds = Collections.singletonMap("topic-1", Uuid.randomUuid());
        metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 5, newTopicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
        assertEquals(Optional.of(5), 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