Sha256: 768fc31eb142b8cef65afe2ebf79902041cae068c4fad439d6c682c1965b2935
Contents?: true
Size: 1.61 KB
Versions: 1
Compression:
Stored size: 1.61 KB
Contents
# encoding: utf-8 require "logstash/devutils/rspec/spec_helper" require "logstash/inputs/kafka" require "digest" require "rspec/wait" 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(:timeout_seconds) { 3600 } let(:num_events) { 103 } def thread_it(kafka_input, queue) Thread.new do begin kafka_input.run(queue) 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(timeout_seconds).for { queue.length }.to eq(num_events) expect(queue.length).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(timeout_seconds).for { queue.length }.to eq(num_events) expect(queue.length).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(timeout_seconds).for { queue.length }.to eq(num_events) expect(queue.length).to eq(num_events) end end
Version data entries
1 entries across 1 versions & 1 rubygems
Version | Path |
---|---|
logstash-input-kafka-5.0.1 | spec/integration/inputs/kafka_spec.rb |