diff --git a/docs/input-kafka.asciidoc b/docs/input-kafka.asciidoc
index 32d29c0..e4408f9 100644
--- a/docs/input-kafka.asciidoc
+++ b/docs/input-kafka.asciidoc
@@ -133,6 +133,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-sasl_client_callback_handler_class>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_oauthbearer_token_endpoint_url>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_oauthbearer_scope_claim_name>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-sasl_login_callback_handler_class>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_login_connect_timeout_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-sasl_login_read_timeout_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-sasl_login_retry_backoff_ms>> |<<number,number>>|No
@@ -583,6 +584,13 @@ The URL for the OAuth 2.0 issuer token endpoint.
 
 (optional) The override name of the scope claim.
 
+[id="plugins-{type}s-{plugin}-sasl_login_callback_handler_class"]
+===== `sasl_login_callback_handler_class`
+  * Value type is <<string,string>>
+  * There is no default value for this setting.
+
+The SASL login callback handler class the specified SASL mechanism should use.
+
 [id="plugins-{type}s-{plugin}-sasl_login_connect_timeout_ms"]
 ===== `sasl_login_connect_timeout_ms`
   * Value type is <<number,number>>
diff --git a/docs/output-kafka.asciidoc b/docs/output-kafka.asciidoc
index 980af65..0a1c53b 100644
--- a/docs/output-kafka.asciidoc
+++ b/docs/output-kafka.asciidoc
@@ -104,6 +104,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-sasl_client_callback_handler_class>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_oauthbearer_token_endpoint_url>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_oauthbearer_scope_claim_name>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-sasl_login_callback_handler_class>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-sasl_login_connect_timeout_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-sasl_login_read_timeout_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-sasl_login_retry_backoff_ms>> |<<number,number>>|No
@@ -419,6 +420,13 @@ The URL for the OAuth 2.0 issuer token endpoint.
 
 (optional) The override name of the scope claim.
 
+[id="plugins-{type}s-{plugin}-sasl_login_callback_handler_class"]
+===== `sasl_login_callback_handler_class`
+  * Value type is <<string,string>>
+  * There is no default value for this setting.
+
+The SASL login callback handler class the specified SASL mechanism should use.
+
 [id="plugins-{type}s-{plugin}-sasl_login_connect_timeout_ms"]
 ===== `sasl_login_connect_timeout_ms`
   * Value type is <<number,number>>
diff --git a/lib/logstash/inputs/kafka.rb b/lib/logstash/inputs/kafka.rb
index 82d235e..4dca75f 100644
--- a/lib/logstash/inputs/kafka.rb
+++ b/lib/logstash/inputs/kafka.rb
@@ -214,6 +214,8 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   config :sasl_oauthbearer_token_endpoint_url, :validate => :string
   # (optional) The override name of the scope claim.
   config :sasl_oauthbearer_scope_claim_name, :validate => :string, :default => 'scope'
+  # SASL login callback handler class
+  config :sasl_login_callback_handler_class, :validate => :string
   # (optional) The duration, in milliseconds, for HTTPS connect timeout
   config :sasl_login_connect_timeout_ms, :validate => :number, :default => 10000
   # (optional) The duration, in milliseconds, for HTTPS read timeout.
diff --git a/lib/logstash/outputs/kafka.rb b/lib/logstash/outputs/kafka.rb
index 842a9da..b2e3ba5 100644
--- a/lib/logstash/outputs/kafka.rb
+++ b/lib/logstash/outputs/kafka.rb
@@ -153,6 +153,8 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
   config :sasl_oauthbearer_token_endpoint_url, :validate => :string
   # (optional) The override name of the scope claim.
   config :sasl_oauthbearer_scope_claim_name, :validate => :string, :default => 'scope'
+  # SASL login callback handler class
+  config :sasl_login_callback_handler_class, :validate => :string
   # (optional) The duration, in milliseconds, for HTTPS connect timeout
   config :sasl_login_connect_timeout_ms, :validate => :number, :default => 10000
   # (optional) The duration, in milliseconds, for HTTPS read timeout.
diff --git a/lib/logstash/plugin_mixins/kafka/common.rb b/lib/logstash/plugin_mixins/kafka/common.rb
index 291519b..6564e80 100644
--- a/lib/logstash/plugin_mixins/kafka/common.rb
+++ b/lib/logstash/plugin_mixins/kafka/common.rb
@@ -44,6 +44,7 @@ def set_sasl_config(props)
       props.put("sasl.client.callback.handler.class", sasl_client_callback_handler_class) unless sasl_client_callback_handler_class.nil?
       props.put("sasl.oauthbearer.token.endpoint.url", sasl_oauthbearer_token_endpoint_url) unless sasl_oauthbearer_token_endpoint_url.nil?
       props.put("sasl.oauthbearer.scope.claim.name", sasl_oauthbearer_scope_claim_name) unless sasl_oauthbearer_scope_claim_name.nil?
+      props.put("sasl.login.callback.handler.class", sasl_login_callback_handler_class) unless sasl_login_callback_handler_class.nil?
       props.put("sasl.login.connect.timeout.ms", sasl_login_connect_timeout_ms.to_s) unless sasl_login_connect_timeout_ms.nil?
       props.put("sasl.login.read.timeout.ms", sasl_login_read_timeout_ms.to_s) unless sasl_login_read_timeout_ms.nil?
       props.put("sasl.login.retry.backoff.ms", sasl_login_retry_backoff_ms.to_s) unless sasl_login_retry_backoff_ms.nil?