@Test
    public void testMetadataMerge() {
        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 topic variants "old" and "keep". Both will be retained.
        String oldClusterId = "oldClusterId";
        int oldNodes = 2;
        Map<String, Errors> oldTopicErrors = new HashMap<>();
        oldTopicErrors.put("oldInvalidTopic", Errors.INVALID_TOPIC_EXCEPTION);
        oldTopicErrors.put("keepInvalidTopic", Errors.INVALID_TOPIC_EXCEPTION);
        oldTopicErrors.put("oldUnauthorizedTopic", Errors.TOPIC_AUTHORIZATION_FAILED);
        oldTopicErrors.put("keepUnauthorizedTopic", Errors.TOPIC_AUTHORIZATION_FAILED);
        Map<String, Integer> oldTopicPartitionCounts = new HashMap<>();
        oldTopicPartitionCounts.put("oldValidTopic", 2);
        oldTopicPartitionCounts.put("keepValidTopic", 3);

        retainTopics.set(Utils.mkSet(
            "oldInvalidTopic",
            "keepInvalidTopic",
            "oldUnauthorizedTopic",
            "keepUnauthorizedTopic",
            "oldValidTopic",
            "keepValidTopic"));

        topicIds.put("oldValidTopic", Uuid.randomUuid());
        topicIds.put("keepValidTopic", Uuid.randomUuid());
        MetadataResponse metadataResponse =
                RequestTestUtils.metadataUpdateWithIds(oldClusterId, oldNodes, oldTopicErrors, oldTopicPartitionCounts, _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)));

        // Update the metadata to add a new topic variant, "new", which will be retained with "keep". Note this
        // means that all of the "old" topics should be dropped.
        Cluster cluster = metadata.fetch();
        assertEquals(cluster.clusterResource().clusterId(), oldClusterId);
        assertEquals(cluster.nodes().size(), oldNodes);
        assertEquals(cluster.invalidTopics(), new HashSet<>(Arrays.asList("oldInvalidTopic", "keepInvalidTopic")));
        assertEquals(cluster.unauthorizedTopics(), new HashSet<>(Arrays.asList("oldUnauthorizedTopic", "keepUnauthorizedTopic")));
        assertEquals(cluster.topics(), new HashSet<>(Arrays.asList("oldValidTopic", "keepValidTopic")));
        assertEquals(cluster.partitionsForTopic("oldValidTopic").size(), 2);
        assertEquals(cluster.partitionsForTopic("keepValidTopic").size(), 3);
        assertEquals(new HashSet<>(cluster.topicIds()), new HashSet<>(topicIds.values()));

        String newClusterId = "newClusterId";
        int newNodes = oldNodes + 1;
        Map<String, Errors> newTopicErrors = new HashMap<>();
        newTopicErrors.put("newInvalidTopic", Errors.INVALID_TOPIC_EXCEPTION);
        newTopicErrors.put("newUnauthorizedTopic", Errors.TOPIC_AUTHORIZATION_FAILED);
        Map<String, Integer> newTopicPartitionCounts = new HashMap<>();
        newTopicPartitionCounts.put("keepValidTopic", 2);
        newTopicPartitionCounts.put("newValidTopic", 4);

        retainTopics.set(Utils.mkSet(
            "keepInvalidTopic",
            "newInvalidTopic",
            "keepUnauthorizedTopic",
            "newUnauthorizedTopic",
            "keepValidTopic",
            "newValidTopic"));

        topicIds.put("newValidTopic", Uuid.randomUuid());
        metadataResponse = RequestTestUtils.metadataUpdateWithIds(newClusterId, newNodes, newTopicErrors, newTopicPartitionCounts, _tp -> 200, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, true, time.milliseconds());
        topicIds.remove("oldValidTopic");
        Map<String, Uuid> metadataTopicIds2 = metadata.topicIds();
        retainTopics.get().forEach(topic -> assertEquals(metadataTopicIds2.get(topic), topicIds.get(topic)));
        assertNull(metadataTopicIds2.get("oldValidTopic"));

        cluster = metadata.fetch();
        assertEquals(cluster.clusterResource().clusterId(), newClusterId);
        assertEquals(cluster.nodes().size(), newNodes);
        assertEquals(cluster.invalidTopics(), new HashSet<>(Arrays.asList("keepInvalidTopic", "newInvalidTopic")));
        assertEquals(cluster.unauthorizedTopics(), new HashSet<>(Arrays.asList("keepUnauthorizedTopic", "newUnauthorizedTopic")));
        assertEquals(cluster.topics(), new HashSet<>(Arrays.asList("keepValidTopic", "newValidTopic")));
        assertEquals(cluster.partitionsForTopic("keepValidTopic").size(), 2);
        assertEquals(cluster.partitionsForTopic("newValidTopic").size(), 4);
        assertEquals(new HashSet<>(cluster.topicIds()), new HashSet<>(topicIds.values()));

        // Perform another metadata update, but this time all topic metadata should be cleared.
        retainTopics.set(Collections.emptySet());

        metadataResponse = RequestTestUtils.metadataUpdateWithIds(newClusterId, newNodes, newTopicErrors, newTopicPartitionCounts, _tp -> 300, topicIds);
        metadata.updateWithCurrentRequestVersion(metadataResponse, true, time.milliseconds());
        Map<String, Uuid> metadataTopicIds3 = metadata.topicIds();
        topicIds.forEach((topicName, topicId) -> assertNull(metadataTopicIds3.get(topicName)));

        cluster = metadata.fetch();
        assertEquals(cluster.clusterResource().clusterId(), newClusterId);
        assertEquals(cluster.nodes().size(), newNodes);
        assertEquals(cluster.invalidTopics(), Collections.emptySet());
        assertEquals(cluster.unauthorizedTopics(), Collections.emptySet());
        assertEquals(cluster.topics(), Collections.emptySet());
        assertTrue(cluster.topicIds().isEmpty());
    }
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