in_kafka.rb in fluent-plugin-kafka-0.6.0

- old
+ new

@@ -44,10 +44,11 @@
                :desc => "How long to block until the server sends us data."
   config_param :min_bytes, :integer, :default => nil,
                :desc => "Smallest amount of data the server should send us."
 
   include Fluent::KafkaPluginUtil::SSLSettings
+  include Fluent::KafkaPluginUtil::SaslSettings
 
   unless method_defined?(:router)
     define_method("router") { Fluent::Engine }
   end
 
@@ -157,10 +158,11 @@
     opt[:min_bytes] = @min_bytes if @min_bytes
 
     @kafka = Kafka.new(seed_brokers: @brokers, client_id: @client_id,
                        ssl_ca_cert: read_ssl_file(@ssl_ca_cert),
                        ssl_client_cert: read_ssl_file(@ssl_client_cert),
-                       ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key))
+                       ssl_client_cert_key: read_ssl_file(@ssl_client_cert_key),
+                       sasl_gssapi_principal: @principal, sasl_gssapi_keytab: @keytab)
     @zookeeper = Zookeeper.new(@offset_zookeeper) if @offset_zookeeper
 
     @topic_watchers = @topic_list.map {|topic_entry|
       offset_manager = OffsetManager.new(topic_entry, @zookeeper, @offset_zk_root_node) if @offset_zookeeper
       TopicWatcher.new(