code: def create_topics(self, new_topics, timeout_ms=None, validate_only=False):
          
          version = self._matching_api_version(CreateTopicsRequest)
          timeout_ms = self._validate_timeout(timeout_ms)
          if version == 0:
              if validate_only:
                  raise IncompatibleBrokerVersion(
                      "validate_only requires CreateTopicsRequest >= v1, which is not supported by Kafka {}."
                      .format(self.config['api_version']))
              request = CreateTopicsRequest[version](
                  create_topic_requests=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
                  timeout=timeout_ms
              )
          elif version <= 2:
              request = CreateTopicsRequest[version](
                  create_topic_requests=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
                  timeout=timeout_ms,
                  validate_only=validate_only
              )
          else:
              raise NotImplementedError(
                  "Support for CreateTopics v{} has not yet been added to KafkaAdminClient."
                  .format(version))
          # TODO convert structs to a more pythonic interface
          # TODO raise exceptions if errors
          return self._send_request_to_controller(request)
docs: Create new topics in the cluster.
  
          :param new_topics: A list of NewTopic objects.
          :param timeout_ms: Milliseconds to wait for new topics to be created
              before the broker returns.
          :param validate_only: If True, don't actually create new topics.
              Not supported by all versions. Default: False
          :return: Appropriate version of CreateTopicResponse class.