def destination(topic, key = nil)
      topic_metadata = topic_metadatas[topic]
      if topic_metadata && topic_metadata.leader_available?
        partition_id = determine_partition(topic_metadata, key)
        broker_id    = topic_metadata.partition_leader(partition_id) || NO_BROKER
      else
        partition_id  = NO_PARTITION
        broker_id     = NO_BROKER
      end

      return partition_id, broker_id
    end
def destination(topic, key = nil)
      topic_metadata = topic_metadatas[topic]
      if topic_metadata && topic_metadata.leader_available?
        partition_id = determine_partition(topic_metadata, key)
        broker_id    = topic_metadata.partition_leader(partition_id) || NO_BROKER
      else
        partition_id  = NO_PARTITION
        broker_id     = NO_BROKER
      end

      return partition_id, broker_id
    end
Create a new message conductor

 @param [Hash<String,TopicMetadata>] topics_metadata
   Metadata for all topics this conductor may receive.
 @param [Object] partitioner
   Custom partitioner
 Determines which partition a message should be sent to.

 @param [String] topic
   Topic we are sending this message to

 @param [Object] key
   Key for this message, may be nil

 @return [Integer,Integer]
   partition_id and broker_id to which this message should be sent