def consumer
      opaque = Opaque.new
      config = native_config(opaque)

      if @consumer_rebalance_listener
        opaque.consumer_rebalance_listener = @consumer_rebalance_listener
        Rdkafka::Bindings.rd_kafka_conf_set_rebalance_cb(config, Rdkafka::Bindings::RebalanceCallback)
      end

      kafka = native_kafka(config, :rd_kafka_consumer)

      # Redirect the main queue to the consumer
      Rdkafka::Bindings.rd_kafka_poll_set_consumer(kafka)

      # Return consumer with Kafka client
      Rdkafka::Consumer.new(kafka)
    end
def consumer
      opaque = Opaque.new
      config = native_config(opaque)

      if @consumer_rebalance_listener
        opaque.consumer_rebalance_listener = @consumer_rebalance_listener
        Rdkafka::Bindings.rd_kafka_conf_set_rebalance_cb(config, Rdkafka::Bindings::RebalanceCallback)
      end

      kafka = native_kafka(config, :rd_kafka_consumer)

      # Redirect the main queue to the consumer
      Rdkafka::Bindings.rd_kafka_poll_set_consumer(kafka)

      # Return consumer with Kafka client
      Rdkafka::Consumer.new(kafka)
    end
Create a consumer with this configuration.

 @raise [ConfigError] When the configuration contains invalid options
 @raise [ClientCreationError] When the native client cannot be created

 @return [Consumer] The created consumer