Sha256: 8bc12467a328b010f80286316197cf366a51a56d3d259cf3be05060a60fc4160
Contents?: true
Size: 1.68 KB
Versions: 8
Compression:
Stored size: 1.68 KB
Contents
# encoding: utf-8
require "logstash/devutils/rspec/spec_helper"
require "logstash/inputs/kafka"
require "digest"
describe "input/kafka", :integration => true do
let(:partition3_config) { { 'topics' => ['topic3'], 'codec' => 'plain', 'auto_offset_reset' => 'earliest'} }
let(:snappy_config) { { 'topics' => ['snappy_topic'], 'codec' => 'plain', 'auto_offset_reset' => 'earliest'} }
let(:lz4_config) { { 'topics' => ['lz4_topic'], 'codec' => 'plain', 'auto_offset_reset' => 'earliest'} }
let(:tries) { 60 }
let(:num_events) { 103 }
def thread_it(kafka_input, queue)
Thread.new do
begin
kafka_input.run(queue)
end
end
end
def wait_for_events(queue, num_events)
begin
timeout(30) do
until queue.length == num_events do
sleep 1
next
end
end
end
end
it "should consume all messages from 3-partition topic" do
kafka_input = LogStash::Inputs::Kafka.new(partition3_config)
queue = Array.new
t = thread_it(kafka_input, queue)
t.run
wait_for_events(queue, num_events)
expect(queue.size).to eq(num_events)
end
it "should consume all messages from snappy 3-partition topic" do
kafka_input = LogStash::Inputs::Kafka.new(snappy_config)
queue = Array.new
t = thread_it(kafka_input, queue)
t.run
wait_for_events(queue, num_events)
expect(queue.size).to eq(num_events)
end
it "should consume all messages from lz4 3-partition topic" do
kafka_input = LogStash::Inputs::Kafka.new(lz4_config)
queue = Array.new
t = thread_it(kafka_input, queue)
t.run
wait_for_events(queue, num_events)
expect(queue.size).to eq(num_events)
end
end
Version data entries
8 entries across 8 versions & 1 rubygems