Added support for the custom client and login callback handler for the kafka-output

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Next
   - [DOC] Updates description of `enable_auto_commit=false` to clarify that the commit happens after data is fetched AND written to the queue [#90](https://github.com/logstash-plugins/logstash-integration-kafka/pull/90)
+  - [Doc] Added support for the custom client and login callback handler for the kafka-output
 
 ## 10.8.1
   - [DOC] Removed a setting recommendation that is no longer applicable for Kafka 2.0+ [#99](https://github.com/logstash-plugins/logstash-integration-kafka/pull/99)

lib/logstash/outputs/kafka.rb

@@ -179,6 +179,10 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
   config :sasl_jaas_config, :validate => :string
   # Optional path to kerberos config file. This is krb5.conf style as detailed in https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html
   config :kerberos_config, :validate => :path
+  # Custom sasl client callback handler class.
+  config :sasl_client_callback_handler_class, :validate => :string
+  # Custom sasl login callback handler class.
+  config :sasl_login_callback_handler_class, :validate => :string
 
   # The topic to produce messages to
   config :topic_id, :validate => :string, :required => true
@@ -362,6 +366,8 @@ def create_producer
       elsif security_protocol == "SASL_PLAINTEXT"
         set_sasl_config(props)
       elsif security_protocol == "SASL_SSL"
+        props.put("sasl.client.callback.handler.class",sasl_client_callback_handler_class) unless sasl_client_callback_handler_class.nil?
+        props.put("sasl.login.callback.handler.class",sasl_login_callback_handler_class) unless sasl_login_callback_handler_class.nil?
         set_trustore_keystore_config(props)
         set_sasl_config(props)
       end