public synchronized Set<TopicPartition> updatePartitionLeadership(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
        Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
        // Insert non-overlapping nodes from existing-nodes into new-nodes.
        this.metadataSnapshot.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));

        // Create partition-metadata for all updated partitions. Exclude updates for partitions -
        // 1. for which the corresponding partition has newer leader in existing metadata.
        // 2. for which corresponding leader's node is missing in the new-nodes.
        // 3. for which the existing metadata doesn't know about the partition.
        List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
        for (Entry<TopicPartition, Metadata.LeaderIdAndEpoch> partitionLeader: partitionLeaders.entrySet()) {
            TopicPartition partition = partitionLeader.getKey();
            Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
            Metadata.LeaderIdAndEpoch newLeader = partitionLeader.getValue();
            if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
                log.debug("For {}, incoming leader information is incomplete {}", partition, newLeader);
                continue;
            }
            if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
                log.debug("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
                continue;
            }
            if (!newNodes.containsKey(newLeader.leaderId.get())) {
                log.debug("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
                continue;
            }
            if (!this.metadataSnapshot.partitionMetadata(partition).isPresent()) {
                log.debug("For {}, incoming leader({}), partition metadata is no longer cached, ignoring.", partition, newLeader);
                continue;
            }

            MetadataResponse.PartitionMetadata existingMetadata = this.metadataSnapshot.partitionMetadata(partition).get();
            MetadataResponse.PartitionMetadata updatedMetadata = new MetadataResponse.PartitionMetadata(
                existingMetadata.error,
                partition,
                newLeader.leaderId,
                newLeader.epoch,
                existingMetadata.replicaIds,
                existingMetadata.inSyncReplicaIds,
                existingMetadata.offlineReplicaIds
            );
            updatePartitionMetadata.add(updatedMetadata);

            lastSeenLeaderEpochs.put(partition, newLeader.epoch.get());
        }

        if (updatePartitionMetadata.isEmpty()) {
            log.debug("No relevant metadata updates.");
            return new HashSet<>();
        }

        Set<String> updatedTopics = updatePartitionMetadata.stream().map(MetadataResponse.PartitionMetadata::topic).collect(Collectors.toSet());

        // Get topic-ids for updated topics from existing topic-ids.
        Map<String, Uuid> existingTopicIds = this.metadataSnapshot.topicIds();
        Map<String, Uuid> topicIdsForUpdatedTopics = updatedTopics.stream()
            .filter(existingTopicIds::containsKey)
            .collect(Collectors.toMap(e -> e, existingTopicIds::get));

        if (log.isDebugEnabled()) {
            updatePartitionMetadata.forEach(
                partMetadata -> log.debug("For {} updating leader information, updated metadata is {}.", partMetadata.topicPartition, partMetadata)
            );
        }

        // Fetch responses can include partition level leader changes, when this happens, we perform a partial
        // metadata update, by keeping the unchanged partition and update the changed partitions.
        this.metadataSnapshot = metadataSnapshot.mergeWith(
            metadataSnapshot.clusterResource().clusterId(),
            newNodes,
            updatePartitionMetadata,
            Collections.emptySet(), Collections.emptySet(), Collections.emptySet(),
            metadataSnapshot.cluster().controller(),
            topicIdsForUpdatedTopics,
            (topic, isInternal) -> true);
        clusterResourceListeners.onUpdate(metadataSnapshot.clusterResource());

        return updatePartitionMetadata.stream()
            .map(metadata -> metadata.topicPartition)
            .collect(Collectors.toSet());
    }
@Test
    public void testUpdatePartitionLeadership() {
        Time time = new MockTime();

        // Initialize metadata
        int numNodes = 5;
        metadata = new Metadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, new LogContext(), new ClusterResourceListeners());
        ClusterResourceListener mockListener = Mockito.mock(ClusterResourceListener.class);
        metadata.addClusterUpdateListener(mockListener);
        // topic1 has 2 partitions: tp11, tp12
        // topic2 has 1 partition: tp21
        String topic1 = "topic1";
        TopicPartition tp11 = new TopicPartition(topic1, 0);
        PartitionMetadata part1Metadata = new PartitionMetadata(Errors.NONE, tp11, Optional.of(1), Optional.of(100), Arrays.asList(1, 2), Arrays.asList(1, 2), Collections.singletonList(3));
        Uuid topic1Id = Uuid.randomUuid();
        TopicPartition tp12 = new TopicPartition(topic1, 1);
        PartitionMetadata part12Metadata = new PartitionMetadata(Errors.NONE, tp12, Optional.of(2), Optional.of(200), Arrays.asList(2, 3), Arrays.asList(2, 3), Collections.singletonList(1));

        String topic2 = "topic2";
        TopicPartition tp21 = new TopicPartition(topic2, 0);
        PartitionMetadata part2Metadata = new PartitionMetadata(Errors.NONE, tp21, Optional.of(2), Optional.of(200), Arrays.asList(2, 3), Arrays.asList(2, 3), Collections.singletonList(1));
        Uuid topic2Id = Uuid.randomUuid();

        Set<String> internalTopics = Collections.singleton(Topic.GROUP_METADATA_TOPIC_NAME);
        TopicPartition internalPart = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0);
        Uuid internalTopicId = Uuid.randomUuid();
        PartitionMetadata internalTopicMetadata = new PartitionMetadata(Errors.NONE, internalPart, Optional.of(2), Optional.of(200), Arrays.asList(2, 3), Arrays.asList(2, 3), Collections.singletonList(1));

        Map<String, Uuid> topicIds = new HashMap<>();
        topicIds.put(topic1, topic1Id);
        topicIds.put(topic2, topic2Id);
        topicIds.put(internalTopics.iterator().next(), internalTopicId);

        Map<String, Integer> topicPartitionCounts = new HashMap<>();
        topicPartitionCounts.put(topic1, 2);
        topicPartitionCounts.put(topic2, 1);
        topicPartitionCounts.put(internalTopics.iterator().next(), 1);
        PartitionMetadataSupplier metadataSupplier = (error, partition, leaderId, leaderEpoch, replicas, isr, offlineReplicas) -> {
            if (partition.equals(tp11))
                return part1Metadata;
            else if (partition.equals(tp21))
                return part2Metadata;
            else if (partition.equals(tp12))
                return part12Metadata;
            else if (partition.equals(internalPart))
                return internalTopicMetadata;
            throw new RuntimeException("Unexpected partition " + partition);
        };

        // Setup invalid topics and unauthorized topics.
        Map<String, Errors> errorCounts = new HashMap<>();
        Set<String> invalidTopics = Collections.singleton("topic3");
        errorCounts.put(invalidTopics.iterator().next(), Errors.INVALID_TOPIC_EXCEPTION);
        Set<String> unauthorizedTopics = Collections.singleton("topic4");
        errorCounts.put(unauthorizedTopics.iterator().next(), Errors.TOPIC_AUTHORIZATION_FAILED);

        metadata.requestUpdate(true);
        Metadata.MetadataRequestAndVersion versionAndBuilder = metadata.newMetadataRequestAndVersion(time.milliseconds());
        assertFalse(versionAndBuilder.isPartialUpdate);
        String clusterId = "kafka-cluster";
        metadata.update(versionAndBuilder.requestVersion,
            RequestTestUtils.metadataUpdateWith(clusterId, numNodes, errorCounts, topicPartitionCounts, tp -> null, metadataSupplier, ApiKeys.METADATA.latestVersion(), topicIds),
            false, time.milliseconds());
        List<Node> nodes = new ArrayList<>(metadata.fetch().nodes());
        Node controller = metadata.fetch().controller();
        assertEquals(numNodes, nodes.size());
        assertFalse(metadata.updateRequested());
        validateForUpdatePartitionLeadership(metadata, part1Metadata, part2Metadata, part12Metadata, internalTopicMetadata, nodes, clusterId, unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds);
        // Since cluster metadata was updated, listener should be called.
        verify(mockListener, times(1)).onUpdate(any());
        Mockito.reset(mockListener);

        // TEST1: Ensure invalid updates get ignored
        Map<TopicPartition, Metadata.LeaderIdAndEpoch> updates = new HashMap<>();
        // New leader info is empty/invalid.
        updates.put(new TopicPartition(topic1, 999), new Metadata.LeaderIdAndEpoch(Optional.empty(), Optional.empty()));
        // Leader's node is unknown
        updates.put(tp21, new  Metadata.LeaderIdAndEpoch(Optional.of(99999), Optional.of(99999)));
        // Partition missing from existing metadata
        updates.put(new TopicPartition("topic_missing_from_existing_metadata", 1), new  Metadata.LeaderIdAndEpoch(Optional.of(0), Optional.of(99999)));
        // New leader info is stale.
        updates.put(tp11, new  Metadata.LeaderIdAndEpoch(part1Metadata.leaderId, Optional.of(part1Metadata.leaderEpoch.get() - 1)));
        Set<TopicPartition> updatedTps = metadata.updatePartitionLeadership(updates, nodes);
        assertTrue(updatedTps.isEmpty());
        validateForUpdatePartitionLeadership(metadata, part1Metadata, part2Metadata, part12Metadata, internalTopicMetadata, nodes, clusterId, unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds);
        // Since cluster metadata is unchanged, listener shouldn't be called.
        verify(mockListener, never()).onUpdate(any());
        Mockito.reset(mockListener);


        //TEST2: Ensure valid update to tp11 is applied to the metadata.  Rest (tp12, tp21) remain unchanged.
        // 1. New Node with id=999 is added.
        // 2. Existing node with id=0 has host, port changed, so is updated.
        Integer part1NewLeaderId = part1Metadata.leaderId.get() + 1;
        Integer part1NewLeaderEpoch = part1Metadata.leaderEpoch.get() + 1;
        updates.put(tp11, new Metadata.LeaderIdAndEpoch(Optional.of(part1NewLeaderId), Optional.of(part1NewLeaderEpoch)));
        PartitionMetadata updatedPart1Metadata = new PartitionMetadata(part1Metadata.error, part1Metadata.topicPartition, Optional.of(part1NewLeaderId), Optional.of(part1NewLeaderEpoch), part1Metadata.replicaIds, part1Metadata.inSyncReplicaIds, part1Metadata.offlineReplicaIds);

        Node newNode = new Node(999, "testhost", 99999, "testrack");
        nodes.add(newNode);
        int index = nodes.stream().filter(node -> node.id() == 0).findFirst().map(nodes::indexOf).orElse(-1);
        Node existingNode = nodes.get(index);
        Node updatedNode = new Node(existingNode.id(), "newhost", existingNode.port(), "newrack");
        nodes.remove(index);
        nodes.add(updatedNode);

        updatedTps = metadata.updatePartitionLeadership(updates, nodes);

        assertEquals(1, updatedTps.size());
        assertEquals(part1Metadata.topicPartition, updatedTps.toArray()[0]);
        // Validate metadata is changed for partition1, hosts are updated, everything else remains unchanged.
        validateForUpdatePartitionLeadership(metadata, updatedPart1Metadata, part2Metadata, part12Metadata, internalTopicMetadata, nodes, clusterId, unauthorizedTopics, invalidTopics, internalTopics, controller, topicIds);
        // Since cluster metadata was updated, listener should be called.
        verify(mockListener, times(1)).onUpdate(any());
        Mockito.reset(mockListener);
    }