diff --git a/.gitignore b/.gitignore
index 85914be..c2d0bca 100644
--- a/.gitignore
+++ b/.gitignore
@@ -8,3 +8,4 @@ venv
 molecule/shared/certs/*
 !molecule/shared/certs/configs/
 !molecule/shared/certs/Makefile
+.vscode
diff --git a/molecule/cluster/molecule.yml b/molecule/cluster/molecule.yml
index 30a0d52..e94f47d 100644
--- a/molecule/cluster/molecule.yml
+++ b/molecule/cluster/molecule.yml
@@ -32,7 +32,6 @@ platforms:
     networks:
       - name: molecule
 
-
   - name: machine3
     image: registry.access.redhat.com/ubi8/ubi-init:latest
     pre_build_image: true
@@ -55,7 +54,6 @@ provisioner:
   inventory:
     group_vars:
       all:
-        kafka_home: /opt/kafka/kafka_2.13-3.1.1
         kafka_advertised_interface_name: eth0
         zookeeper_interface_name: eth0
       
@@ -71,8 +69,11 @@ provisioner:
 
         kafka_tls_config:
           enabled: true
-          trustedCA: /etc/certs/ca-root.pem
+          trustedCA:
+            file: ../../../molecule/shared/certs/ca-root.pem
+            location: /etc/certs/ca-root.pem
           keystore:
+            file: ../../../molecule/shared/certs/cert-cluster.p12
             location: /etc/certs/cert-cluster.p12
             password: changeit
             type: PKCS12
@@ -106,12 +107,17 @@ provisioner:
         kafka_admin:
           listener_name: admin
           authentication:
-            tls:
-              trustedCA: /etc/certs/ca-root.pem
-              keystore:
-                location: /etc/certs/cert-cluster.p12
-                password: changeit
-                type: PKCS12
+            type: tls
+          tls:
+            enabled: true
+            trustedCA:
+              file: ../../../molecule/shared/certs/ca-root.pem
+              location: /etc/certs/ca-root.pem
+            keystore:
+              file: ../../../molecule/shared/certs/cert-admin.p12
+              location: /etc/certs/cert-admin.p12
+              password: changeit
+              type: PKCS12
 
     host_vars:
       machine1:
diff --git a/molecule/zk-secured/converge.yml b/molecule/zk-secured/converge.yml
new file mode 100644
index 0000000..99ab811
--- /dev/null
+++ b/molecule/zk-secured/converge.yml
@@ -0,0 +1,2 @@
+---
+- import_playbook: ../../playbooks/setup.yml
\ No newline at end of file
diff --git a/molecule/zk-secured/molecule.yml b/molecule/zk-secured/molecule.yml
new file mode 100644
index 0000000..dddff23
--- /dev/null
+++ b/molecule/zk-secured/molecule.yml
@@ -0,0 +1,113 @@
+---
+# skip prerun because I am experiencing the issue described here:
+# https://github.com/ansible/ansible-lint/issues/2070
+prerun: false
+dependency:
+  name: galaxy
+driver:
+  name: docker
+lint: |
+  ansible-lint -v
+platforms:
+
+  - name: machine1
+    image: registry.access.redhat.com/ubi8/ubi-init:latest
+    pre_build_image: true
+    privileged: true
+    command: "/usr/sbin/init"
+    groups:
+      - kafka
+      - zookeeper
+    networks:
+      - name: molecule
+
+  - name: machine2
+    image: registry.access.redhat.com/ubi8/ubi-init:latest
+    pre_build_image: true
+    privileged: true
+    command: "/usr/sbin/init"
+    groups:
+      - kafka
+      - zookeeper
+    networks:
+      - name: molecule
+
+  - name: machine3
+    image: registry.access.redhat.com/ubi8/ubi-init:latest
+    pre_build_image: true
+    privileged: true
+    command: "/usr/sbin/init"
+    groups:
+      - kafka
+      - zookeeper
+    networks:
+      - name: molecule
+
+provisioner:
+  name: ansible
+  ansible_args:
+    - --diff
+  playbooks:
+    converge: converge.yml
+    prepare: ../shared/prepare.yml
+    verify: verify.yml
+  inventory:
+    group_vars:
+      zookeeper:
+        zookeeper_interface_name: eth0
+
+        zookeeper_auth_client_enabled: true
+        zookeeper_auth_quorum_enabled: true
+        zookeeper_auth_type: tls
+
+        zookeeper_tls_config:
+          enabled: true
+          trustedCA:
+            file: ../../../molecule/shared/certs/ca-root.pem
+            location: /etc/certs/ca-root.pem
+          keystore:
+            file: ../../../molecule/shared/certs/cert-cluster.p12
+            location: /etc/certs/cert-cluster.p12
+            password: changeit
+            type: PKCS12
+
+      kafka:
+        kafka_advertised_interface_name: eth0
+        kafka_advertised_host: "{{ ansible_hostname }}"
+        
+        kafka_tls_config:
+          enabled: true
+          trustedCA:
+            file: ../../../molecule/shared/certs/ca-root.pem
+            location: /etc/certs/ca-root.pem
+          keystore:
+            file: ../../../molecule/shared/certs/cert-cluster.p12
+            location: /etc/certs/cert-cluster.p12
+            password: changeit
+            type: PKCS12
+
+        kafka_listeners:
+
+          - name: plain
+            port: 9092
+            tls: false
+
+        kafka_inter_broker_listener_name: plain
+        
+        kafka_admin:
+          listener_name: plain
+
+    host_vars:
+      machine1:
+        kafka_broker_id: 1
+        zookeeper_myid: 1
+      machine2:
+        kafka_broker_id: 2
+        zookeeper_myid: 2
+      machine3:
+        kafka_broker_id: 3
+        zookeeper_myid: 3
+      localhost:
+        ansible_python_interpreter: "{{ ansible_playbook_python }}"
+verifier:
+  name: ansible
diff --git a/molecule/zk-secured/verify.yml b/molecule/zk-secured/verify.yml
new file mode 100644
index 0000000..3a45222
--- /dev/null
+++ b/molecule/zk-secured/verify.yml
@@ -0,0 +1,47 @@
+---
+- name: Verify
+  hosts: all
+  gather_facts: true
+  vars: 
+    kafka_home: /opt/kafka/kafka_2.13-3.7.0
+  tasks:
+  
+  - name: Ensure ZK CLI config file exist
+    ansible.builtin.copy:
+      content: |
+        zookeeper.ssl.keystore.location=/etc/certs/cert-cluster.p12
+        zookeeper.ssl.keystore.password=changeit
+        zookeeper.ssl.keystore.type=PKCS12
+        zookeeper.ssl.truststore.location=/etc/certs/ca-root.pem
+        zookeeper.ssl.truststore.type=PEM
+        zookeeper.ssl.client.enable=true
+        zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
+      dest: /tmp/zk-tls-config.properties
+  
+  - name: Test ZK CLI mTLS connection
+    ansible.builtin.command: >
+      {{ kafka_home }}/bin/zookeeper-shell.sh \
+        {{ ansible_default_ipv4.address }}:2182 \
+        whoami \
+        -zk-tls-config-file /tmp/zk-tls-config.properties
+    changed_when: false
+    register: zk_cli_res
+
+  - name: ZK CLI mTLS connection assertions
+    ansible.builtin.assert:
+      that: 
+        - zk_cli_res.stdout_lines is search('SyncConnected') 
+        - zk_cli_res.stdout_lines is search('x509:\ CN=kafka')
+
+  - name: Test Kafka to ZK mTLS connection
+    ansible.builtin.command: >
+      {{ kafka_home }}/bin/kafka-topics.sh \
+        --bootstrap-server {{ ansible_default_ipv4.address }}:9092 \
+        --describe
+    changed_when: false
+    run_once: true
+    register: kafka_zk_res
+
+  - name: Kafka to ZK mTLS connection assertions
+    ansible.builtin.assert:
+      that: kafka_zk_res.rc == 0
\ No newline at end of file
diff --git a/plugins/action/kafka_config.py b/plugins/action/kafka_config.py
index cb24677..188c2f0 100644
--- a/plugins/action/kafka_config.py
+++ b/plugins/action/kafka_config.py
@@ -17,13 +17,18 @@
     keystore=dict(
       type='dict',
       options=dict(
+        file=dict(type='str'),
         location=dict(type='str'),
         password=dict(type='str'),
         type=dict(type='str')
       )
     ),
     trustedCA=dict(
-      type='str'
+      type='dict',
+      options=dict(
+        file=dict(type='str'),
+        location=dict(type='str')
+      )
     )
   )
 )
@@ -110,7 +115,7 @@ def run(self, tmp=None, task_vars=None):
     ret = dict()
     ret['__kafka_config'] = kafka_config
     ret['__admin_config'] = generator.get_admin_config(kafka_config)
-    
+
 
     return dict(ansible_facts=dict(ret))
 
@@ -152,10 +157,8 @@ def _coalesce_listeners(self, listeners):
         listener['protocol'] = 'PLAINTEXT'
 
       if tls:
-        # TODO support different trustore types
         listener['truststore_type'] = 'PEM'
-        listener['truststore_location'] = tls['trustedCA']
-        # listener['truststore_password'] = 
+        listener['truststore_location'] = tls['trustedCA']['location']
         
         listener['keystore_location'] = tls['keystore']['location']
         listener['keystore_password'] = tls['keystore']['password']
@@ -246,10 +249,10 @@ def get_admin_config(self, kafka_config):
     options = {}
     
     authentication = admin.pop('authentication', {})
-    tls = authentication.pop('tls', {})
+    tls = admin.pop('tls', {})
     
     if 'trustedCA' in tls:
-      options['ssl.truststore.location']=tls['trustedCA']
+      options['ssl.truststore.location']=tls['trustedCA']['location']
       options['ssl.truststore.type']='PEM'
    
     if 'keystore' in tls:
@@ -291,7 +294,5 @@ def get_admin_config(self, kafka_config):
       'protocol': protocol,
       'listener': '%s:%s' % (listener['advertised'], listener['port']),
       'require_command_config': bool(options),
-      'options': options 
+      'options': options
     }
-
-
diff --git a/roles/kafka_systemd_broker/tasks/setup_admin.yml b/roles/kafka_systemd_broker/tasks/setup_admin.yml
index a62da0d..eacff81 100644
--- a/roles/kafka_systemd_broker/tasks/setup_admin.yml
+++ b/roles/kafka_systemd_broker/tasks/setup_admin.yml
@@ -10,4 +10,3 @@
     - __admin_config.require_command_config is defined
     - __admin_config.require_command_config
   become: true
-    
\ No newline at end of file
diff --git a/roles/kafka_systemd_broker/tasks/setup_security.yml b/roles/kafka_systemd_broker/tasks/setup_security.yml
index 934ff18..e130c73 100644
--- a/roles/kafka_systemd_broker/tasks/setup_security.yml
+++ b/roles/kafka_systemd_broker/tasks/setup_security.yml
@@ -4,7 +4,22 @@
     kafka_java_opts: "{{ kafka_java_opts + [
         '-Djava.security.auth.login.config=' + kafka_jaas_config_file_location
       ] }}"
-  when: zookeeper_auth_type is defined
+  when: 
+    - zookeeper_auth_client_enabled
+    - zookeeper_auth_type is defined
+    - zookeeper_auth_type == 'gssapi'
+
+- name: Set SSL facts
+  ansible.builtin.set_fact:
+    kafka_ssl_truststore_file: "{{ kafka_tls_config.trustedCA.file }}"
+    kafka_ssl_truststore_location: "{{ kafka_tls_config.trustedCA.location }}"
+    kafka_ssl_keystore_file: "{{ kafka_tls_config.keystore.file }}"
+    kafka_ssl_keystore_location: "{{ kafka_tls_config.keystore.location }}"
+    kafka_ssl_keystore_password: "{{ kafka_tls_config.keystore.password }}"
+    kafka_ssl_keystore_type: "{{ kafka_tls_config.keystore.type }}"
+  when: 
+  - kafka_tls_config is defined
+  - kafka_tls_config.enabled
 
 - name: Create security dir for Kafka
   ansible.builtin.file:
@@ -22,12 +37,15 @@
     owner: "{{ kafka_user }}"
     group: "{{ kafka_group }}"
     mode: 0400
-  when: zookeeper_auth_type is defined
+  when: 
+    - zookeeper_auth_client_enabled
+    - zookeeper_auth_type is defined
+    - zookeeper_auth_type == 'gssapi'
   become: true
 
 - name: Copy SSL resources
   block:
-    - name: Copy java truststore file
+    - name: Copy truststore file
       ansible.builtin.copy:
         src: "{{ kafka_ssl_truststore_file }}"
         dest: "{{ kafka_ssl_truststore_location }}"
@@ -35,7 +53,7 @@
         group: "{{ kafka_group }}"
         mode: 0400
 
-    - name: Copy java keystore file
+    - name: Copy keystore file
       ansible.builtin.copy:
         src: "{{ kafka_ssl_keystore_file }}"
         dest: "{{ kafka_ssl_keystore_location }}"
@@ -56,5 +74,5 @@
       mode: 0400
   when:
     - item.value.authentication.type is defined
-    - item.value.authentication.type in 'gssapi'
+    - item.value.authentication.type == 'gssapi'
   become: true
\ No newline at end of file
diff --git a/roles/kafka_systemd_broker/templates/server.properties.j2 b/roles/kafka_systemd_broker/templates/server.properties.j2
index 3bce7d9..07f921e 100644
--- a/roles/kafka_systemd_broker/templates/server.properties.j2
+++ b/roles/kafka_systemd_broker/templates/server.properties.j2
@@ -11,11 +11,20 @@ inter.broker.listener.name={{ kafka_inter_broker_listener_name }}
 log.dirs={{ kafka_log_dirs | join(',') }}
 
 zookeeper.connect={{ kafka_zookeeper_connect_servers | join(',') }}
-# TODO allow to configure zookeeper
+{% if zookeeper_auth_client_enabled and zookeeper_auth_type == 'tls' %}
+zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
+zookeeper.ssl.client.enable=true
+zookeeper.ssl.keystore.location={{ kafka_ssl_keystore_location }}
+zookeeper.ssl.keystore.password={{ kafka_ssl_keystore_password }}
+zookeeper.ssl.keystore.type={{ kafka_ssl_keystore_type }}
+zookeeper.ssl.truststore.location={{ kafka_ssl_truststore_location }}
+zookeeper.ssl.truststore.type=PEM
+zookeeper.set.acl=true
+{% endif %}
 
 {% for l_name, l_config in __kafka_config.listeners.items() %}
 
-{% if l_config.tls %}
+{%- if l_config.tls -%}
 # ---- {{ l_name }} -----
 listener.name.{{ l_name | lower }}.ssl.truststore.location={{ l_config.truststore_location }}
 listener.name.{{ l_name | lower }}.ssl.truststore.type={{ l_config.truststore_type }}
@@ -25,7 +34,7 @@ listener.name.{{ l_name | lower }}.ssl.keystore.type={{ l_config.keystore_type }
 {% if 'ssl_client_auth' in l_config %}
 listener.name.{{ l_name | lower }}.ssl.client.auth={{ l_config.ssl_client_auth }}
 {% endif %}
-{% endif %}
+{%- endif -%}
 
 {% if 'sasl' in l_config %}
 listener.name.{{ l_name | lower }}.sasl.enabled.mechanisms={{ l_config.sasl.keys() | map('upper') | join(',') }}
@@ -42,7 +51,7 @@ sasl.mechanism.inter.broker.protocol={{ l_config.authentication.type | upper }}
 
 {% endfor %}
 
-{% if __kafka_config.authorization %}
+{%- if __kafka_config.authorization -%}
 allow.everyone.if.no.acl.found = {{ __kafka_config.authorization['allow.everyone.if.no.acl.found'] }}
 authorizer.class.name = {{ __kafka_config.authorization['authorizer.class.name'] }}
 super.users = {{ __kafka_config.authorization['super.users'] | join(';') }}
@@ -50,5 +59,4 @@ super.users = {{ __kafka_config.authorization['super.users'] | join(';') }}
 
 {% for kafka_conf_key, kafka_conf_value in __kafka_config.additional_config.items() %}
 {{ kafka_conf_key }}={{ kafka_conf_value }}
-{% endfor %}
-
+{% endfor %}
\ No newline at end of file
diff --git a/roles/kafka_systemd_zookeeper/defaults/main.yml b/roles/kafka_systemd_zookeeper/defaults/main.yml
index 5ffa2ab..ef233ab 100644
--- a/roles/kafka_systemd_zookeeper/defaults/main.yml
+++ b/roles/kafka_systemd_zookeeper/defaults/main.yml
@@ -16,6 +16,7 @@ zookeeper_interface_name: eth1
 
 zookeeper_port_offset: 0
 zookeeper_client_port: "{{ 2181 + (zookeeper_port_offset | int) }}"
+zookeeper_secure_client_port: "{{ 2182 + (zookeeper_port_offset | int) }}"
 zookeeper_client_address: "{{ ansible_default_ipv4.address | default(ansible_all_ipv4_addresses[0]) }}"
 
 zookeeper_servers_ansible_group: zookeeper
diff --git a/roles/kafka_systemd_zookeeper/tasks/main.yml b/roles/kafka_systemd_zookeeper/tasks/main.yml
index a4a77fc..c8733f5 100644
--- a/roles/kafka_systemd_zookeeper/tasks/main.yml
+++ b/roles/kafka_systemd_zookeeper/tasks/main.yml
@@ -63,7 +63,7 @@
 # ---------------------------
 - name: Setup security resources
   ansible.builtin.include_tasks: setup_security.yml
-  when: zookeeper_auth_quorum_enabled
+  when: zookeeper_auth_type is defined
   
 - name: Set Zookeeper Servers Facts
   include_tasks: set_zk_facts.yml
diff --git a/roles/kafka_systemd_zookeeper/tasks/service_check.yml b/roles/kafka_systemd_zookeeper/tasks/service_check.yml
index 05dfc70..ad810d8 100644
--- a/roles/kafka_systemd_zookeeper/tasks/service_check.yml
+++ b/roles/kafka_systemd_zookeeper/tasks/service_check.yml
@@ -2,5 +2,5 @@
 - name: Verify Zookeeper is listening
   ansible.builtin.wait_for:
     host: "{{ zookeeper_client_address }}"
-    port: "{{ zookeeper_client_port }}"
+    port: "{{ zookeeper_tls_enabled | ternary(zookeeper_secure_client_port, zookeeper_client_port) }}"
     timeout: 60
diff --git a/roles/kafka_systemd_zookeeper/tasks/set_zk_facts.yml b/roles/kafka_systemd_zookeeper/tasks/set_zk_facts.yml
index 13aac69..3cb6753 100644
--- a/roles/kafka_systemd_zookeeper/tasks/set_zk_facts.yml
+++ b/roles/kafka_systemd_zookeeper/tasks/set_zk_facts.yml
@@ -7,7 +7,8 @@
 
 - name: Init Zookeeper servers facts
   ansible.builtin.set_fact:
-    zookeeper_address_fqdn: zookeeper_auth_type is defined and zookeeper_auth_type in 'gssapi'
+    zookeeper_gssapi_enabled: "{{ zookeeper_auth_type is defined and zookeeper_auth_type == 'gssapi' }}"
+    zookeeper_tls_enabled: "{{ zookeeper_auth_type is defined and zookeeper_auth_type == 'tls' }}"
     zookeeper_servers: []
 
 - name: Print Zookeeper server address
@@ -19,8 +20,8 @@
   ansible.builtin.set_fact:
     zookeeper_servers: "{{ zookeeper_servers + [{
           'myid': hostvars[item].zookeeper_myid,
-          'address': zookeeper_address_fqdn | ternary(hostvars[item]['ansible_facts']['fqdn'], hostvars[item].ansible_facts[zookeeper_interface_name].ipv4.address),
-          'client_port': 2181 + (zookeeper_port_offset | int),
+          'address': zookeeper_gssapi_enabled | ternary(hostvars[item]['ansible_facts']['fqdn'], hostvars[item].ansible_facts[zookeeper_interface_name].ipv4.address),
+          'client_port': zookeeper_tls_enabled | ternary(zookeeper_secure_client_port, zookeeper_client_port),
           'follower_port': 2888 + (zookeeper_port_offset | int),
           'election_port': 3888 + (zookeeper_port_offset | int)
         }]
diff --git a/roles/kafka_systemd_zookeeper/tasks/setup_security.yml b/roles/kafka_systemd_zookeeper/tasks/setup_security.yml
index 3cbcb87..93d7178 100644
--- a/roles/kafka_systemd_zookeeper/tasks/setup_security.yml
+++ b/roles/kafka_systemd_zookeeper/tasks/setup_security.yml
@@ -4,7 +4,21 @@
     zookeeper_java_opts: "{{ zookeeper_java_opts + [
         '-Djava.security.auth.login.config=' + zookeeper_jaas_config_file_location
       ] }}"
-  when: zookeeper_auth_type is defined
+  when: 
+    - zookeeper_auth_type is defined
+    - zookeeper_auth_type == 'gssapi'
+
+- name: Set SSL facts
+  ansible.builtin.set_fact:
+    zookeeper_ssl_truststore_file: "{{ zookeeper_tls_config.trustedCA.file }}"
+    zookeeper_ssl_truststore_location: "{{ zookeeper_tls_config.trustedCA.location }}"
+    zookeeper_ssl_keystore_file: "{{ zookeeper_tls_config.keystore.file }}"
+    zookeeper_ssl_keystore_location: "{{ zookeeper_tls_config.keystore.location }}"
+    zookeeper_ssl_keystore_password: "{{ zookeeper_tls_config.keystore.password }}"
+    zookeeper_ssl_keystore_type: "{{ zookeeper_tls_config.keystore.type }}"
+  when: 
+  - zookeeper_tls_config is defined
+  - zookeeper_tls_config.enabled
 
 - name: Create security dir for Kafka
   ansible.builtin.file:
@@ -25,7 +39,27 @@
     mode: 0400
   when:
     - zookeeper_auth_type is defined 
-    - zookeeper_auth_type in 'gssapi'
+    - zookeeper_auth_type == 'gssapi'
+  become: true
+
+- name: Copy SSL resources
+  block:
+    - name: Copy truststore file
+      copy:
+        src: "{{ zookeeper_ssl_truststore_file }}"
+        dest: "{{ zookeeper_ssl_truststore_location }}"
+        owner: "{{ zookeeper_user }}"
+        group: "{{ zookeeper_group }}"
+
+    - name: Copy keystore file
+      copy:
+        src: "{{ zookeeper_ssl_keystore_file }}"
+        dest: "{{ zookeeper_ssl_keystore_location }}"
+        owner: "{{ zookeeper_user }}"
+        group: "{{ zookeeper_group }}"
+  when:
+    - zookeeper_auth_type is defined 
+    - zookeeper_auth_type == 'tls'
   become: true
 
 - name: Copy jaas config file
@@ -37,5 +71,5 @@
     mode: 0400
   when:
     - zookeeper_auth_type is defined 
-    - zookeeper_auth_type in 'gssapi'
+    - zookeeper_auth_type == 'gssapi'
   become: true
\ No newline at end of file
diff --git a/roles/kafka_systemd_zookeeper/templates/zookeeper.properties.j2 b/roles/kafka_systemd_zookeeper/templates/zookeeper.properties.j2
index bd10db8..c40a17e 100644
--- a/roles/kafka_systemd_zookeeper/templates/zookeeper.properties.j2
+++ b/roles/kafka_systemd_zookeeper/templates/zookeeper.properties.j2
@@ -2,6 +2,10 @@
 
 dataDir={{ zookeeper_data_path }}
 clientPort={{ zookeeper_client_port }}
+{% if zookeeper_auth_type is defined and 'tls' in zookeeper_auth_type %}
+secureClientPort={{ zookeeper_secure_client_port }}
+serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
+{% endif %}
 
 tickTime=2000
 initLimit=10
@@ -20,13 +24,22 @@ server.{{ zookeeper_server.myid }}={{ zookeeper_server.address }}:{{ zookeeper_s
 
 {% if zookeeper_auth_quorum_enabled %}
 # Server-to-Server authentication
+{% if 'tls' in zookeeper_auth_type %}
+sslQuorum=true
+ssl.quorum.clientAuth=need
+ssl.quorum.keyStore.location={{ zookeeper_ssl_keystore_location }}
+ssl.quorum.keyStore.password={{ zookeeper_ssl_keystore_password }}
+ssl.quorum.keyStore.type=PKCS12
+ssl.quorum.trustStore.location={{ zookeeper_ssl_truststore_location }}
+ssl.quorum.trustStore.type=PEM
+{% elif zookeeper_conf_quorum_enable_sasl %}
 quorum.auth.enableSasl={{ zookeeper_conf_quorum_enable_sasl | ternary('true', 'false') }}
 quorum.auth.learnerRequireSasl={{ zookeeper_conf_quorum_learner_require_sasl | ternary('true', 'false') }}
 quorum.auth.serverRequireSasl={{ zookeeper_conf_quorum_server_require_sasl | ternary('true', 'false') }}
 quorum.auth.learner.loginContext=QuorumLearner
 quorum.auth.server.loginContext=QuorumServer
 quorum.cnxn.threads.size=20
-
+{% endif %}
 {% if 'gssapi' in zookeeper_auth_type %}
 # Can be adjusted to suit ticket renewal intervals. Default is one hour.
 jaasLoginRenew=3600000
@@ -34,14 +47,23 @@ kerberos.removeHostFromPrincipal=false
 kerberos.removeRealmFromPrincipal=false
 quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST
 {% endif %}
-
 {% endif %}
 
-{%- if zookeeper_auth_client_enabled %}
+{% if zookeeper_auth_client_enabled %}
 # Client-to-Server authentication
+{% if 'tls' in zookeeper_auth_type %}
+authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
+ssl.clientAuth=need
+ssl.keyStore.location={{ zookeeper_ssl_keystore_location }}
+ssl.keyStore.password={{ zookeeper_ssl_keystore_password }}
+ssl.keyStore.type=PKCS12
+ssl.trustStore.location={{ zookeeper_ssl_truststore_location }}
+ssl.trustStore.type=PEM
+{% elif zookeeper_conf_quorum_enable_sasl %}
 # You must add the authProvider.<ID> property for every server that is part of the ZooKeeper cluster.
 requireClientAuthScheme=sasl
 {% for zookeeper_server in zookeeper_servers %}
 authProvider.{{ zookeeper_server.myid }}=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
 {% endfor %}
 {% endif %}
+{% endif %}
\ No newline at end of file
diff --git a/tests/plugins/action/test_kafka_config.py b/tests/plugins/action/test_kafka_config.py
index 4c2e6bf..b534387 100644
--- a/tests/plugins/action/test_kafka_config.py
+++ b/tests/plugins/action/test_kafka_config.py
@@ -17,9 +17,12 @@ def yaml_args():
 
 tls:
   enabled: true
-  trustedCA: defaultCA.pem
+  trustedCA:
+    file: ../../../molecule/shared/certs/ca-root.pem
+    location: /etc/certs/ca-root.pem
   keystore:
-    location: cert.p12
+    file: ../../../molecule/shared/certs/cert-cluster.p12
+    location: /etc/certs/cert-cluster.p12
     password: changeit
     type: PKCS12
 
@@ -48,13 +51,17 @@ def yaml_args():
 admin:
   listener_name: admin
   authentication:
-    tls:
-      trustedCA: /ect/certs/ca-root.pem
-      keystore:
-        location: /etc/certs/cert-admin.p12
-        password: changeit
-        type: PKCS12    
-
+    type: tls
+  tls:
+    enabled: true
+    trustedCA:
+      file: ../../../molecule/shared/certs/ca-root.pem
+      location: /etc/certs/ca-root.pem
+    keystore:
+      file: ../../../molecule/shared/certs/cert-admin.p12
+      location: /etc/certs/cert-admin.p12
+      password: changeit
+      type: PKCS12
 """)
 
 def test_convert_empty_configuration():
@@ -66,9 +73,9 @@ def test_convert_empty_configuration():
 
 def test_convert_kafka_config(yaml_args):
     actual = KafkaConfigGenerator(yaml_args).get_kafka_config()
-    assert actual['core']['listeners'] == 'REPLICATION://:9091,AUTHENTICATED://:9094,ADMIN://:9095'
-    assert actual['core']['advertised.listeners'] == 'REPLICATION://localhost:9091,AUTHENTICATED://localhost:9094,ADMIN://localhost:9095'
-    assert actual['core']['listener.security.protocol.map'] == 'REPLICATION:SSL,AUTHENTICATED:SASL_SSL,ADMIN:SSL'
+    assert actual['core']['listeners'] == 'replication://:9091,authenticated://:9094,admin://:9095'
+    assert actual['core']['advertised.listeners'] == 'replication://localhost:9091,authenticated://localhost:9094,admin://localhost:9095'
+    assert actual['core']['listener.security.protocol.map'] == 'replication:SSL,authenticated:SASL_SSL,admin:SSL'
     assert actual['authorization']['authorizer.class.name'] == 'kafka.security.authorizer.AclAuthorizer'
     assert actual['listeners']['authenticated']['sasl']['scram-sha-512'] == 'org.apache.kafka.common.security.scram.ScramLoginModule required;'
 
@@ -76,14 +83,11 @@ def test_convert_admin_configuration(yaml_args):
     kafka_config_generator = KafkaConfigGenerator(yaml_args)
     kafka_config = kafka_config_generator.get_kafka_config()
     actual = kafka_config_generator.get_admin_config(kafka_config)
-    print(actual)
     assert actual['listener'] == 'localhost:9095'
-    assert actual['options']['ssl.truststore.location'] == '/ect/certs/ca-root.pem'
+    assert actual['options']['ssl.truststore.location'] == '/etc/certs/ca-root.pem'
     assert actual['options']['ssl.truststore.type'] == 'PEM'
     assert actual['options']['ssl.keystore.location'] == '/etc/certs/cert-admin.p12'
     assert actual['options']['ssl.keystore.type'] == 'PKCS12'
     assert actual['options']['ssl.keystore.password'] == 'changeit'
-  
-    
-
     
+  
\ No newline at end of file