@Test
    public void testMetadataMergeOnIdDowngrade() {
        Time time = new MockTime();
        Map<String, Uuid> topicIds = new HashMap<>();

        final AtomicReference<Set<String>> retainTopics = new AtomicReference<>(new HashSet<>());
        metadata = new Metadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, new LogContext(), new ClusterResourceListeners()) {
            @Override
            protected boolean retainTopic(String topic, boolean isInternal, long nowMs) {
                return retainTopics.get().contains(topic);
            }
        };

        // Initialize a metadata instance with two topics. Both will be retained.
        String clusterId = "clusterId";
        int nodes = 2;
        Map<String, Integer> topicPartitionCounts = new HashMap<>();
        topicPartitionCounts.put("validTopic1", 2);
        topicPartitionCounts.put("validTopic2", 3);

        retainTopics.set(Utils.mkSet(
                "validTopic1",
                "validTopic2"));

        topicIds.put("validTopic1", Uuid.randomUuid());
        topicIds.put("validTopic2", Uuid.randomUuid());
        MetadataResponse metadataResponse =
                RequestTestUtils.metadataUpdateWithIds(clusterId, nodes, Collections.emptyMap(), topicPartitionCounts, _tp -> 100, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, true, time.milliseconds());
        Map<String, Uuid> metadataTopicIds1 = metadata.topicIds();
        retainTopics.get().forEach(topic -> assertEquals(metadataTopicIds1.get(topic), topicIds.get(topic)));

        // Try removing the topic ID from keepValidTopic (simulating receiving a request from a controller with an older IBP)
        topicIds.remove("validTopic1");
        metadataResponse = RequestTestUtils.metadataUpdateWithIds(clusterId, nodes, Collections.emptyMap(), topicPartitionCounts, _tp -> 200, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, true, time.milliseconds());
        Map<String, Uuid> metadataTopicIds2 = metadata.topicIds();
        retainTopics.get().forEach(topic -> assertEquals(metadataTopicIds2.get(topic), topicIds.get(topic)));

        Cluster cluster = metadata.fetch();
        // We still have the topic, but it just doesn't have an ID.
        assertEquals(Utils.mkSet("validTopic1", "validTopic2"), cluster.topics());
        assertEquals(2, cluster.partitionsForTopic("validTopic1").size());
        assertEquals(new HashSet<>(topicIds.values()), new HashSet<>(cluster.topicIds()));
        assertEquals(Uuid.ZERO_UUID, cluster.topicId("validTopic1"));
    }
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