From f2f6a976121705c43dbd065ff27b7d68b11a131d Mon Sep 17 00:00:00 2001
From: Karen Metts <35154725+karenzone@users.noreply.github.com>
Date: Wed, 14 Jul 2021 10:38:29 -0400
Subject: [PATCH] Doc: Update changelog with info/link to PR #90 (#100)

---
 CHANGELOG.md                  | 1 +
 lib/logstash/outputs/kafka.rb | 6 ++++++
 2 files changed, 7 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9c3a4aca..94b2539a 100644
--- a/CHANGELOG.md
+++ b/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)
diff --git a/lib/logstash/outputs/kafka.rb b/lib/logstash/outputs/kafka.rb
index b7f4c629..39f101ba 100644
--- a/lib/logstash/outputs/kafka.rb
+++ b/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 client 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