diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2d1ce82..b76541c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 7.3.1
+  - Added support for kafka property `ssl.endpoint.identification.algorithm` [#213](https://github.com/logstash-plugins/logstash-output-kafka/pull/213)
+
 ## 7.3.0
   - Changed Kafka client to version 2.1.0
 
diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 20142a3..a32bf0a 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -82,6 +82,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<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
 | <<plugins-{type}s-{plugin}-send_buffer_bytes>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-ssl_endpoint_identification_algorithm>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ssl_key_password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-ssl_keystore_location>> |a valid filesystem path|No
 | <<plugins-{type}s-{plugin}-ssl_keystore_password>> |<<password,password>>|No
@@ -332,6 +333,14 @@ Security protocol to use, which can be either of PLAINTEXT,SSL,SASL_PLAINTEXT,SA
 
 The size of the TCP send buffer to use when sending data.
 
+[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}-ssl_key_password"]
 ===== `ssl_key_password` 
 
diff --git a/lib/logstash/outputs/kafka.rb b/lib/logstash/outputs/kafka.rb
index 6cd29a0..1abda11 100644
--- a/lib/logstash/outputs/kafka.rb
+++ b/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, :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"
   # 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) unless ssl_endpoint_identification_algorithm.nil?
   end
 
   def set_sasl_config(props)
diff --git a/logstash-output-kafka.gemspec b/logstash-output-kafka.gemspec
index a154862..9076554 100644
--- a/logstash-output-kafka.gemspec
+++ b/logstash-output-kafka.gemspec
@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-output-kafka'
-  s.version         = '7.3.0'
+  s.version         = '7.3.1'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Writes events to a Kafka topic"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"