def subscribe(*topics)
      # Create topic partition list with topics and no partition set
      tpl = TopicPartitionList.new_native_tpl(topics.length)

      topics.each do |topic|
        Rdkafka::Bindings.rd_kafka_topic_partition_list_add(
          tpl,
          topic,
          -1
        )
      end
      # Subscribe to topic partition list and check this was successful
      response = Rdkafka::Bindings.rd_kafka_subscribe(@native_kafka, tpl)
      if response != 0
        raise Rdkafka::RdkafkaError.new(response, "Error subscribing to '#{topics.join(', ')}'")
      end
    end
def subscribe(*topics)
      # Create topic partition list with topics and no partition set
      tpl = TopicPartitionList.new_native_tpl(topics.length)

      topics.each do |topic|
        Rdkafka::Bindings.rd_kafka_topic_partition_list_add(
          tpl,
          topic,
          -1
        )
      end
      # Subscribe to topic partition list and check this was successful
      response = Rdkafka::Bindings.rd_kafka_subscribe(@native_kafka, tpl)
      if response != 0
        raise Rdkafka::RdkafkaError.new(response, "Error subscribing to '#{topics.join(', ')}'")
      end
    end
Subscribe to one or more topics letting Kafka handle partition assignments.

 @param topics [Array<String>] One or more topic names

 @raise [RdkafkaError] When subscribing fails

 @return [nil]