def perform_leader_election(self, election_type, topic_partitions=None, timeout_ms=None):
        """Perform leader election on the topic partitions.

        :param election_type: Type of election to attempt. 0 for Perferred, 1 for Unclean
        :param topic_partitions: A map of topic name strings to partition ids list.
            By default, will run on all topic partitions
        :param timeout_ms: Milliseconds to wait for the leader election process to complete
            before the broker returns.

        :return: Appropriate version of ElectLeadersResponse class.
        """
        version = self._client.api_version(ElectLeadersRequest, max_version=1)
        timeout_ms = self._validate_timeout(timeout_ms)
        request = ElectLeadersRequest[version](
            election_type=ElectionType(election_type),
            topic_partitions=self._get_topic_partitions(topic_partitions),
            timeout=timeout_ms,
        )
        # TODO convert structs to a more pythonic interface
        return self._send_request_to_controller(request)
Perform leader election on the topic partitions.

        :param election_type: Type of election to attempt. 0 for Perferred, 1 for Unclean
        :param topic_partitions: A map of topic name strings to partition ids list.
            By default, will run on all topic partitions
        :param timeout_ms: Milliseconds to wait for the leader election process to complete
            before the broker returns.

        :return: Appropriate version of ElectLeadersResponse class.
        
kafka/admin/client.py