def _get_leader_for_partitions(self, partitions, timeout_ms=None):
        """Finds ID of the leader node for every given topic partition.

        Will raise UnknownTopicOrPartitionError if for some partition no leader can be found.

        :param partitions: ``[TopicPartition]``: partitions for which to find leaders.
        :param timeout_ms: ``float``: Timeout in milliseconds, if None (default), will be read from
            config.

        :return: Dictionary with ``{leader_id -> {partitions}}``
        """
        timeout_ms = self._validate_timeout(timeout_ms)

        partitions = set(partitions)
        topics = set(tp.topic for tp in partitions)

        response = self._get_cluster_metadata(topics=topics).to_object()

        leader2partitions = defaultdict(list)
        valid_partitions = set()
        for topic in response.get("topics", ()):
            for partition in topic.get("partitions", ()):
                t2p = TopicPartition(topic=topic["topic"], partition=partition["partition"])
                if t2p in partitions:
                    leader2partitions[partition["leader"]].append(t2p)
                    valid_partitions.add(t2p)

        if len(partitions) != len(valid_partitions):
            unknown = set(partitions) - valid_partitions
            raise UnknownTopicOrPartitionError(
                "The following partitions are not known: %s"
                % ", ".join(str(x) for x in unknown)
            )

        return leader2partitions
Finds ID of the leader node for every given topic partition.

        Will raise UnknownTopicOrPartitionError if for some partition no leader can be found.

        :param partitions: ``[TopicPartition]``: partitions for which to find leaders.
        :param timeout_ms: ``float``: Timeout in milliseconds, if None (default), will be read from
            config.

        :return: Dictionary with ``{leader_id -> {partitions}}``
        
kafka/admin/client.py