docs: def _refresh_controller_id(self):
          """Determine the Kafka cluster controller."""
          version = self._matching_api_version(MetadataRequest)
          if 1 <= version <= 6:
              request = MetadataRequest[version]()
              response = self._send_request_to_node(self._client.least_loaded_node(), request)
              controller_id = response.controller_id
              # verify the controller is new enough to support our requests
              controller_version = self._client.check_version(controller_id)
              if controller_version < (0, 10, 0):
                  raise IncompatibleBrokerVersion(
                      "The controller appears to be running Kafka {}. KafkaAdminClient requires brokers >= 0.10.0.0."
                      .format(controller_version))
              self._controller_id = controller_id
          else:
              raise UnrecognizedBrokerVersion(
                  "Kafka Admin interface cannot determine the controller using MetadataRequest_v{}."
                  .format(version))