code: def _next_partition(self, topic, key=None):
          
          # check if the client has metadata for the topic
          while self.client.metadata_error_for_topic(topic):
              # client doesn't have good metadata for topic. ask to fetch...
              # check if we have request attempts left
              if self._req_attempts >= self._max_attempts:
                  # No, no attempts left, so raise the error
                  _check_error(self.client.metadata_error_for_topic(topic))
              yield self.client.load_metadata_for_topics(topic)
              if not self.client.metadata_error_for_topic(topic):
                  break
              self._req_attempts += 1
              d = Deferred()
              self.client.reactor.callLater(
                  self._retry_interval, d.callback, True)
              self._retry_interval *= self.RETRY_INTERVAL_FACTOR
              yield d
  
          # Ok, should be safe to get the partitions now...
          partitions = self.client.topic_partitions[topic]
          # Do we have a partitioner for this topic already?
          if topic not in self.partitioners:
              # No, create a new paritioner for topic, partitions
              self.partitioners[topic] = \
                  self.partitioner_class(topic, partitions)
          # Lookup the next partition
          partition = self.partitioners[topic].partition(key, partitions)
          returnValue(partition)
docs: get the next partition to which to publish
  
          Check with our client for the latest partitions for the topic, then
          ask our partitioner for the next partition to which we should publish
          for the give key. If needed, create a new partitioner for the topic.