added ssl_endpoint_identification_algorithm config option

Fixes #213

docs/index.asciidoc

@@ -89,6 +89,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-ssl_truststore_location>> |a valid filesystem path|No
 | <<plugins-{type}s-{plugin}-ssl_truststore_password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-ssl_truststore_type>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-ssl_endpoint_identification_algorithm>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-topic_id>> |<<string,string>>|Yes
 | <<plugins-{type}s-{plugin}-value_serializer>> |<<string,string>>|No
 |=======================================================================
@@ -388,6 +389,14 @@ The truststore password
 
 The truststore type.
 
+[id="plugins-{type}s-{plugin}-ssl_endpoint_identification_algorithm"]
+===== `ssl_endpoint_identification_algorithm` 
+
+  * Value type is <<string,string>>
+  * Default value is `"https"`
+
+The endpoint identification algorithm, defaults to "https". Set to empty string "" to disable
+
 [id="plugins-{type}s-{plugin}-topic_id"]
 ===== `topic_id` 
 

lib/logstash/outputs/kafka.rb

@@ -140,6 +140,8 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
   config :ssl_keystore_password, :validate => :password
   # The password of the private key in the key store file.
   config :ssl_key_password, :validate => :password
+  # Algorithm to use when verifying host. Set to "" to disable
+  config :ssl_endpoint_identification_algorithm, :validate => :string
   # 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"
   # http://kafka.apache.org/documentation.html#security_sasl[SASL mechanism] used for client connections. 
@@ -365,6 +367,7 @@ def set_trustore_keystore_config(props)
     props.put("ssl.key.password", ssl_key_password.value) unless ssl_key_password.nil?
     props.put("ssl.keystore.location", ssl_keystore_location) unless ssl_keystore_location.nil?
     props.put("ssl.keystore.password", ssl_keystore_password.value) unless ssl_keystore_password.nil?
+    props.put("ssl.endpoint.identification.algorithm", ssl_endpoint_identification_algorithm.value) unless ssl_endpoint_identification_algorithm.nil?
   end
 
   def set_sasl_config(props)