From 7786c5ebdc3bd3092faa5866d21c41bc34629cef Mon Sep 17 00:00:00 2001
From: Andrea Selva <selva.andre@gmail.com>
Date: Mon, 26 Aug 2024 10:25:08 +0200
Subject: [PATCH] Exposed sasl.client.callback.handler.class Kafka client
 setting to select a specific SASL Handler class (#177)

Co-authored-by: Akos Korsos <akos.korsos@n26.com>
---
 docs/input-kafka.asciidoc                  | 8 ++++++++
 docs/output-kafka.asciidoc                 | 8 ++++++++
 lib/logstash/inputs/kafka.rb               | 2 ++
 lib/logstash/outputs/kafka.rb              | 2 ++
 lib/logstash/plugin_mixins/kafka/common.rb | 1 +
 5 files changed, 21 insertions(+)

diff --git a/docs/input-kafka.asciidoc b/docs/input-kafka.asciidoc
index 6ad10691..5c37688a 100644
--- a/docs/input-kafka.asciidoc
+++ b/docs/input-kafka.asciidoc
@@ -130,6 +130,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-reconnect_backoff_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-request_timeout_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-retry_backoff_ms>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-sasl_client_callback_handler_class>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_jaas_config>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_kerberos_service_name>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_mechanism>> |<<string,string>>|No
@@ -555,6 +556,13 @@ retries are exhausted.
 The amount of time to wait before attempting to retry a failed fetch request
 to a given topic partition. This avoids repeated fetching-and-failing in a tight loop.
 
+[id="plugins-{type}s-{plugin}-sasl_client_callback_handler_class""]
+===== `sasl_client_callback_handler_class`
+* Value type is <<string,string>>
+* There is no default value for this setting.
+
+The SASL client callback handler class the specified SASL mechanism should use.
+
 [id="plugins-{type}s-{plugin}-sasl_jaas_config"]
 ===== `sasl_jaas_config` 
 
diff --git a/docs/output-kafka.asciidoc b/docs/output-kafka.asciidoc
index afe0162d..716d3dd6 100644
--- a/docs/output-kafka.asciidoc
+++ b/docs/output-kafka.asciidoc
@@ -101,6 +101,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-request_timeout_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-retries>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-retry_backoff_ms>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-sasl_client_callback_handler_class>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_jaas_config>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_kerberos_service_name>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_mechanism>> |<<string,string>>|No
@@ -391,6 +392,13 @@ In versions prior to 10.5.0, any exception is retried indefinitely unless the `r
 
 The amount of time to wait before attempting to retry a failed produce request to a given topic partition.
 
+[id="plugins-{type}s-{plugin}-sasl_client_callback_handler_class""]
+===== `sasl_client_callback_handler_class`
+* Value type is <<string,string>>
+* There is no default value for this setting.
+
+The SASL client callback handler class the specified SASL mechanism should use.
+
 [id="plugins-{type}s-{plugin}-sasl_jaas_config"]
 ===== `sasl_jaas_config` 
 
diff --git a/lib/logstash/inputs/kafka.rb b/lib/logstash/inputs/kafka.rb
index d730e567..604bc683 100644
--- a/lib/logstash/inputs/kafka.rb
+++ b/lib/logstash/inputs/kafka.rb
@@ -208,6 +208,8 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   config :ssl_endpoint_identification_algorithm, :validate => :string, :default => 'https'
   # Security protocol to use, which can be either of PLAINTEXT,SSL,SASL_PLAINTEXT,SASL_SSL
   config :security_protocol, :validate => ["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"], :default => "PLAINTEXT"
+  # SASL client callback handler class
+  config :sasl_client_callback_handler_class, :validate => :string
   # http://kafka.apache.org/documentation.html#security_sasl[SASL mechanism] used for client connections. 
   # This may be any mechanism for which a security provider is available.
   # GSSAPI is the default mechanism.
diff --git a/lib/logstash/outputs/kafka.rb b/lib/logstash/outputs/kafka.rb
index fee07b7b..ebf233fc 100644
--- a/lib/logstash/outputs/kafka.rb
+++ b/lib/logstash/outputs/kafka.rb
@@ -147,6 +147,8 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
   config :ssl_endpoint_identification_algorithm, :validate => :string, :default => 'https'
   # Security protocol to use, which can be either of PLAINTEXT,SSL,SASL_PLAINTEXT,SASL_SSL
   config :security_protocol, :validate => ["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"], :default => "PLAINTEXT"
+  # SASL client callback handler class
+  config :sasl_client_callback_handler_class, :validate => :string
   # http://kafka.apache.org/documentation.html#security_sasl[SASL mechanism] used for client connections. 
   # This may be any mechanism for which a security provider is available.
   # GSSAPI is the default mechanism.
diff --git a/lib/logstash/plugin_mixins/kafka/common.rb b/lib/logstash/plugin_mixins/kafka/common.rb
index 79e0d4b7..1ae8546e 100644
--- a/lib/logstash/plugin_mixins/kafka/common.rb
+++ b/lib/logstash/plugin_mixins/kafka/common.rb
@@ -41,6 +41,7 @@ def set_sasl_config(props)
 
       props.put("sasl.kerberos.service.name", sasl_kerberos_service_name) unless sasl_kerberos_service_name.nil?
       props.put("sasl.jaas.config", sasl_jaas_config) unless sasl_jaas_config.nil?
+      props.put("sasl.client.callback.handler.class", sasl_client_callback_handler_class) unless sasl_client_callback_handler_class.nil?
     end
 
     def reassign_dns_lookup