Exposing sasl.jaas.config setting

docs/index.asciidoc

@@ -105,6 +105,7 @@ https://kafka.apache.org/documentation for more details.
 | <<plugins-{type}s-{plugin}-reconnect_backoff_ms>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-request_timeout_ms>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-retry_backoff_ms>> |<<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
 | <<plugins-{type}s-{plugin}-security_protocol>> |<<string,string>>, one of `["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"]`|No
@@ -421,6 +422,24 @@ 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_jaas_config"]
+===== `sasl_jaas_config` 
+
+  * Value type is <<string,string>>
+  * There is no default value for this setting.
+
+JAAS configuration setting local to this plugin instance, as opposed to settings using config file configured using `jaas_path`, which are shared across the JVM. This allows each plugin instance to have its own configuration. 
+
+If both `sasl_jaas_config` and `jaas_path` configurations are set, the setting here takes precedence.
+
+Example (setting for Azure Event Hub):
+[source,ruby]
+    input {
+      kafka {
+        sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='auser'  password='apassword';"
+      }
+    }
+
 [id="plugins-{type}s-{plugin}-sasl_kerberos_service_name"]
 ===== `sasl_kerberos_service_name` 
 

lib/logstash/inputs/kafka.rb

@@ -201,6 +201,8 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   # `jaas_path` and `kerberos_config`. If this is not desirable, you would have to run separate instances of Logstash on 
   # different JVM instances.
   config :jaas_path, :validate => :path
+  # JAAS configuration settings. This allows JAAS config to be a part of the plugin configuration and allows for different JAAS configuration per each plugin config.
+  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
   # Option to add Kafka metadata like topic, message size to the event.
@@ -355,5 +357,6 @@ def set_sasl_config(props)
     end
 
     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?
   end
 end #class LogStash::Inputs::Kafka