def _send_request_to_controller(self, request):
        """Send a Kafka protocol message to the cluster controller.

        Will block until the message result is received.

        Arguments:
            request: The message to send.

        Returns:
            The Kafka protocol response for the message.
        """
        tries = 2  # in case our cached self._controller_id is outdated
        while tries:
            tries -= 1
            future = self._send_request_to_node(self._controller_id, request)

            self._wait_for_futures([future])

            response = future.value
            # In Java, the error field name is inconsistent:
            #  - CreateTopicsResponse / CreatePartitionsResponse uses topic_errors
            #  - DeleteTopicsResponse uses topic_error_codes
            # So this is a little brittle in that it assumes all responses have
            # one of these attributes and that they always unpack into
            # (topic, error_code) tuples.
            topic_error_tuples = getattr(response, 'topic_errors', getattr(response, 'topic_error_codes', None))
            if topic_error_tuples is not None:
                success = self._parse_topic_request_response(topic_error_tuples, request, response, tries)
            else:
                # Leader Election request has a two layer error response (topic and partition)
                success = self._parse_topic_partition_request_response(request, response, tries)

            if success:
                return response
        raise RuntimeError("This should never happen, please file a bug with full stacktrace if encountered")
Send a Kafka protocol message to the cluster controller.

        Will block until the message result is received.

        Arguments:
            request: The message to send.

        Returns:
            The Kafka protocol response for the message.
        
kafka/admin/client.py