code: def _send_request_to_controller(self, request):
          
          tries = 2  # in case our cached self._controller_id is outdated
          while tries:
              tries -= 1
              response = self._send_request_to_node(self._controller_id, request)
              # In Java, the error fieldname 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 = (response.topic_errors if hasattr(response, 'topic_errors')
                  else response.topic_error_codes)
              # Also small py2/py3 compatibility -- py3 can ignore extra values
              # during unpack via: for x, y, *rest in list_of_values. py2 cannot.
              # So for now we have to map across the list and explicitly drop any
              # extra values (usually the error_message)
              for topic, error_code in map(lambda e: e[:2], topic_error_tuples):
                  error_type = Errors.for_code(error_code)
                  if tries and error_type is NotControllerError:
                      # No need to inspect the rest of the errors for
                      # non-retriable errors because NotControllerError should
                      # either be thrown for all errors or no errors.
                      self._refresh_controller_id()
                      break
                  elif error_type is not Errors.NoError:
                      raise error_type(
                          "Request '{}' failed with response '{}'."
                          .format(request, response))
              else:
                  return response
          raise RuntimeError("This should never happen, please file a bug with full stacktrace if encountered")
docs: Send a Kafka protocol message to the cluster controller.
  
          Will block until the message result is received.
  
          :param request: The message to send.
          :return: The Kafka protocol response for the message.