Add tests for out_rdkafka2

Signed-off-by: Takuro Ashie <[email protected]>

Author: ashie

fluent-plugin-kafka.gemspec

@@ -19,6 +19,7 @@ Gem::Specification.new do |gem|
   gem.add_dependency "fluentd", [">= 0.10.58", "< 2"]
   gem.add_dependency 'ltsv'
   gem.add_dependency 'ruby-kafka', '>= 1.4.0', '< 2'
+  gem.add_development_dependency "rdkafka", ">= 0.10.0"
   gem.add_development_dependency "rake", ">= 0.9.2"
   gem.add_development_dependency "test-unit", ">= 3.0.8"
   gem.add_development_dependency "test-unit-rr", "~> 1.0"

test/plugin/test_out_rdkafka2.rb

@@ -0,0 +1,107 @@
+require 'helper'
+require 'fluent/test/helpers'
+require 'fluent/test/driver/input'
+require 'fluent/test/driver/output'
+require 'securerandom'
+require 'fluent/plugin/out_rdkafka2'
+
+class Rdkafka2OutputTest < Test::Unit::TestCase
+  include Fluent::Test::Helpers
+
+  def setup
+    Fluent::Test.setup
+  end
+
+  def base_config
+    config_element('ROOT', '', {"@type" => "rdkafka2"}, [
+                     config_element('format', "", {"@type" => "json"})
+                   ])
+  end
+
+  def config(default_topic: "kitagawakeiko")
+    base_config + config_element('ROOT', '', {"default_topic" => default_topic,
+                                              "brokers" => "localhost:9092"}, [
+                                 ])
+  end
+
+  def create_driver(conf = config, tag='test')
+    Fluent::Test::Driver::Output.new(Fluent::Rdkafka2Output).configure(conf)
+  end
+
+  def test_configure
+    assert_nothing_raised(Fluent::ConfigError) {
+      create_driver(base_config)
+    }
+
+    assert_nothing_raised(Fluent::ConfigError) {
+      create_driver(config)
+    }
+
+    assert_nothing_raised(Fluent::ConfigError) {
+      create_driver(config + config_element('buffer', "", {"@type" => "memory"}))
+    }
+
+    d = create_driver
+    assert_equal 'kitagawakeiko', d.instance.default_topic
+    assert_equal 'localhost:9092', d.instance.brokers
+  end
+
+  def test_mutli_worker_support
+    d = create_driver
+    assert_equal true, d.instance.multi_workers_ready?
+  end
+
+  class WriteTest < self
+    TOPIC_NAME = "kafka-output-#{SecureRandom.uuid}"
+
+    INPUT_CONFIG = %[
+      @type kafka
+      brokers localhost:9092
+      format json
+      @label @kafka
+      topics #{TOPIC_NAME}
+    ]
+
+    def create_target_driver(conf = INPUT_CONFIG)
+      Fluent::Test::Driver::Input.new(Fluent::KafkaInput).configure(conf)
+    end
+
+    def setup
+      @kafka = Kafka.new(["localhost:9092"], client_id: 'kafka')
+    end
+
+    def teardown
+      @kafka.delete_topic(TOPIC_NAME)
+      @kafka.close
+    end
+
+    def test_write
+      target_driver = create_target_driver
+      expected_message = {"a" => 2}
+      target_driver.run(expect_records: 1, timeout: 5) do
+        sleep 2
+        d = create_driver(config(default_topic: TOPIC_NAME))
+        d.run do
+          d.feed("test", event_time, expected_message)
+        end
+      end
+      actual_messages = target_driver.events.collect { |event| event[2] }
+      assert_equal([expected_message], actual_messages)
+    end
+
+    def test_exclude_fields
+      conf = config(default_topic: TOPIC_NAME) +
+             config_element('ROOT', '', {"exclude_fields" => "$.foo"}, [])
+      target_driver = create_target_driver
+      target_driver.run(expect_records: 1, timeout: 5) do
+        sleep 2
+        d = create_driver(conf)
+        d.run do
+          d.feed('test', event_time, {'a' => 'b', 'foo' => 'bar', 'message' => 'test'})
+        end
+      end
+      actual_messages = target_driver.events.collect { |event| event[2] }
+      assert_equal([{'a' => 'b', 'message' => 'test'}], actual_messages)
+    end
+  end
+end