require 'integration/multiple_brokers/spec_helper'

RSpec.describe "consuming with multiple brokers", :type => :request do
  before(:each) do
    @tc = ThreeBrokerCluster.new
    @tc.start

    # autocreate the topic by asking for information about it
    c = Connection.new("localhost", 9092, "metadata_fetcher", 10_000)
    md = c.topic_metadata(["test"])
    spec_sleep 1, "sreating topic"
  end

  after(:each) do
    @tc.stop
  end

  it "finds the lead broker for each partition" do
    brokers = Set.new
    0.upto(2) do |partition|
      pc = PartitionConsumer.consumer_for_partition("test_client",
                                                    ["localhost:9092"],
                                                    "test", partition,
                                                    :earliest_offset)

      brokers.add("#{pc.host}:#{pc.port}")
    end
    expect(brokers.size).to eq(3)
  end

  it "consumes from all partitions" do
    @p = Producer.new(["localhost:9092","localhost:9093","localhost:9094"], "test",
                     :required_acks => 1)

    msgs = 24.times.map { |n| "hello_#{n}" }
    msgs.each do |msg|
      @p.send_messages([MessageToSend.new("test", msg)])
    end

    fetched_messages = []
    0.upto(2) do |partition|
      pc = PartitionConsumer.consumer_for_partition("test_client",
                                                    ["localhost:9092"],
                                                    "test", partition,
                                                    :earliest_offset)
      fetched_messages.push(*pc.fetch)
    end
    expect(fetched_messages.map(&:value).sort).to eq(msgs.sort)
  end
end
