diff --git a/elk_metrics_6x/installLogstash.yml b/elk_metrics_6x/installLogstash.yml
index 78d1caff..c6462f27 100644
--- a/elk_metrics_6x/installLogstash.yml
+++ b/elk_metrics_6x/installLogstash.yml
@@ -158,6 +158,29 @@
         - logstash-filters
         - config
 
+    - name: Run kafka output block
+      block:
+        - name: Copy kafka keystore into place
+          copy:
+            src: "{{ logstash_kafka_ssl_keystore_location }}"
+            dest: "/var/lib/logstash/{{ logstash_kafka_ssl_keystore_location | basename }}"
+          when:
+            - logstash_kafka_ssl_keystore_location is defined
+
+        - name: Copy kafka truststore into place
+          copy:
+            src: "{{ logstash_kafka_ssl_truststore_location }}"
+            dest: "/var/lib/logstash/{{ logstash_kafka_ssl_truststore_location | basename }}"
+          when:
+            - logstash_kafka_ssl_truststore_location is defined
+
+        - name: Deploy Logstash kafka configuration files
+          template:
+            src: "templates/99-kafka-output.conf"
+            dest: "/etc/logstash/conf.d/99-kafka-output.conf"
+      when:
+        - logstash_kafka_options is defined
+
     - name: Ensure logstash ownership
       file:
         path: /var/lib/logstash
diff --git a/elk_metrics_6x/readme.rst b/elk_metrics_6x/readme.rst
index 33995fda..73b98b85 100644
--- a/elk_metrics_6x/readme.rst
+++ b/elk_metrics_6x/readme.rst
@@ -398,6 +398,57 @@ auto-generate the tags list with the following command.
     openstack-ansible setup-openstack.yml --tags "$(cat setup-openstack.yml | grep -wo 'os-.*' | awk -F'-' '{print $2 "-config"}' | tr '\n' ',')"
 
 
+Optional | add Kafka Output format
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+To send data from Logstash to Kafka create the `logstash_kafka_options`
+variable. This variable will be used as a generator and create a Kafka output
+configuration file using the key/value pairs as options.
+
+.. code-block:: yaml
+
+    logstash_kafka_options:
+      codec: json
+      topic_id: "elk_kafka"
+      ssl_key_password: "{{ logstash_kafka_ssl_key_password }}"
+      ssl_keystore_password: "{{ logstash_kafka_ssl_keystore_password }}"
+      ssl_keystore_location: "/var/lib/logstash/{{ logstash_kafka_ssl_keystore_location | basename }}"
+      ssl_truststore_location: "/var/lib/logstash/{{ logstash_kafka_ssl_truststore_location | basename }}"
+      ssl_truststore_password: "{{ logstash_kafka_ssl_truststore_password }}"
+      bootstrap_servers:
+        - server1.local:9092
+        - server2.local:9092
+        - server3.local:9092
+      client_id: "elk_metrics_6x"
+      compression_type: "gzip"
+      security_protocol: "SSL"
+
+
+For a complete list of all options available within the Logstash Kafka output
+plugin please review the `following documentation <https://www.elastic.co/guide/en/logstash/current/plugins-outputs-kafka.html>`_.
+
+Optional config:
+  The following variables are optional and correspond to the example
+  `logstash_kafka_options` variable.
+
+.. code-block:: yaml
+
+    logstash_kafka_ssl_key_password: "secrete"
+    logstash_kafka_ssl_keystore_password: "secrete"
+    logstash_kafka_ssl_truststore_password: "secrete"
+
+    # SSL certificates in Java KeyStore format
+    logstash_kafka_ssl_keystore_location: "/root/kafka/keystore.jks"
+    logstash_kafka_ssl_truststore_location: "/root/kafka/truststore.jks"
+
+
+When using the kafka output plugin the options,
+`logstash_kafka_ssl_keystore_location` and
+`logstash_kafka_ssl_truststore_location` will automatically copy a local SSL key
+to the logstash nodes. These options are string value and assume the deployment
+nodes have local access to the files.
+
+
 Optional | add Grafana visualizations
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
diff --git a/elk_metrics_6x/templates/99-kafka-output.conf.j2 b/elk_metrics_6x/templates/99-kafka-output.conf.j2
new file mode 100644
index 00000000..315b4d2a
--- /dev/null
+++ b/elk_metrics_6x/templates/99-kafka-output.conf.j2
@@ -0,0 +1,13 @@
+output {
+  kafka {
+{% for key, value in logstash_kafka_options.items() %}
+{%   if value is number %}
+    {{ key }} => {{ value }}
+{%   elif value is iterable and value is not string %}
+    {{ key }} => "{{ value | join(',') }}"
+{%   else %}
+    {{ key }} => "{{ value }}"
+{%   endif %}
+{% endfor %}
+  }
+}
diff --git a/elk_metrics_6x/vars/variables.yml b/elk_metrics_6x/vars/variables.yml
index d4dbbac1..0a300ae7 100644
--- a/elk_metrics_6x/vars/variables.yml
+++ b/elk_metrics_6x/vars/variables.yml
@@ -41,6 +41,32 @@ kibana_server_name: "{{ ansible_hostname }}"
 logstash_beat_input_port: 5044
 logstash_deploy_filters: true
 
+## Logstash config showing a complete kafka setup using SSL for authentication.
+# logstash_kafka_options:
+#   codec: json
+#   topic_id: "elk_kafka"
+#   ssl_key_password: "{{ logstash_kafka_ssl_key_password }}"
+#   ssl_keystore_password: "{{ logstash_kafka_ssl_keystore_password }}"
+#   ssl_keystore_location: "/var/lib/logstash/{{ logstash_kafka_ssl_keystore_location | basename }}"
+#   ssl_truststore_location: "/var/lib/logstash/{{ logstash_kafka_ssl_truststore_location | basename }}"
+#   ssl_truststore_password: "{{ logstash_kafka_ssl_truststore_password }}"
+#   bootstrap_servers:
+#     - server1.local:9092
+#     - server2.local:9092
+#     - server3.local:9092
+#   client_id: "elk_metrics_6x"
+#   compression_type: "gzip"
+#   security_protocol: "SSL"
+
+## The following variables are options that correspond to the
+## `logstash_kafka_options` variable.
+# logstash_kafka_ssl_key_password: "secrete"
+# logstash_kafka_ssl_keystore_password: "secrete"
+# logstash_kafka_ssl_truststore_password: "secrete"
+# logstash_kafka_ssl_keystore_location: "/root/kafka/keystore.jks"
+# logstash_kafka_ssl_truststore_location: "/root/kafka/truststore.jks"
+
+
 # APM vars
 apm_interface: 0.0.0.0
 apm_port: 8200