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());
    }
Updates the partition-leadership info in the metadata. Update is done by merging existing metadata with the input leader information and nodes.
This is called whenever partition-leadership updates are returned in a response from broker(ex - ProduceResponse & FetchResponse).
Note that the updates via Metadata RPC are handled separately in ({@link #update}).
Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
@param partitionLeaders map of new leadership information for partitions.
@param leaderNodes a list of nodes for leaders in the above map.
@return a set of partitions, for which leaders were updated.
clients/src/main/java/org/apache/kafka/clients/Metadata.java