diff --git a/.vscode/settings.json b/.vscode/settings.json
new file mode 100644
index 000000000..7d30d567a
--- /dev/null
+++ b/.vscode/settings.json
@@ -0,0 +1,3 @@
+{
+    "ansible.python.interpreterPath": "/opt/homebrew/bin/python3.12"
+}
\ No newline at end of file
diff --git a/roles/kafka_controller/tasks/health_check.yml b/roles/kafka_controller/tasks/health_check.yml
index eee86c01d..dfa6f819c 100644
--- a/roles/kafka_controller/tasks/health_check.yml
+++ b/roles/kafka_controller/tasks/health_check.yml
@@ -9,6 +9,23 @@
   ignore_errors: false
   changed_when: false
   check_mode: false
+  when:
+    - not kafka_controller_client_secrets_protection_enabled|bool
+
+
+- name: Check Kafka Metadata Quorum with Secrets Protection enabled
+  shell: |
+    {{ binary_base_path }}/bin/kafka-metadata-quorum --bootstrap-server {{inventory_hostname}}:{{kafka_controller_port}} \
+      --command-config {{kafka_controller.client_config_file}} describe --replication
+  environment:
+    CONFLUENT_SECURITY_MASTER_KEY: "{{ secrets_protection_masterkey }}"
+    KAFKA_OPTS: "-Xlog:all=error -XX:+IgnoreUnrecognizedVMOptions {% if kerberos_client_config_file_dest != '/etc/krb5.conf' %}-Djava.security.krb5.conf={{kerberos_client_config_file_dest}}{% endif %}"
+  ignore_errors: false
+  changed_when: false
+  check_mode: false
+  when:
+    - kafka_controller_client_secrets_protection_enabled|bool
+
 
 #Registers LEO of controllers only if Metadata Quorum passed in the above task
 - name: Register LogEndOffset
@@ -21,6 +38,23 @@
   ignore_errors: false
   changed_when: false
   check_mode: false
+  when:
+    - not kafka_controller_client_secrets_protection_enabled|bool
+
+#Registers LEO of controllers only if Metadata Quorum passed in the above task
+- name: Register LogEndOffset with Secrets Protection enabled
+  shell: |
+    {{ binary_base_path }}/bin/kafka-metadata-quorum --bootstrap-server {{inventory_hostname}}:{{kafka_controller_port}} \
+      --command-config {{kafka_controller.client_config_file}} describe --replication |  grep -v Observer | awk '{print $2}'
+  environment:
+    CONFLUENT_SECURITY_MASTER_KEY: "{{ secrets_protection_masterkey }}"
+    KAFKA_OPTS: "-Xlog:all=error -XX:+IgnoreUnrecognizedVMOptions {% if kerberos_client_config_file_dest != '/etc/krb5.conf' %}-Djava.security.krb5.conf={{kerberos_client_config_file_dest}}{% endif %}"
+  register: LEO_SECRETS_PROTECTION
+  ignore_errors: false
+  changed_when: false
+  check_mode: false
+  when:
+    - kafka_controller_client_secrets_protection_enabled|bool
 
 - name: Check LogEndOffset values
   assert:
@@ -32,12 +66,27 @@
   ignore_errors: false
   changed_when: false
   check_mode: false
+  when:
+    - not kafka_controller_client_secrets_protection_enabled|bool
+
+- name: Check LogEndOffset values with Secrets Protection enabled
+  assert:
+    that:
+      - "{{ item|int > 0 and LEO_SECRETS_PROTECTION.stdout_lines[1:]|max|int - item|int < 1000 }}"
+    fail_msg: "UnreachableQuorumMember or Found at least one quorum voter with an offset {{ item }}, while the primary controller was at offset {{ LEO_SECRETS_PROTECTION.stdout_lines[1:]|max}}
+               The max allowed offset lag is 1000"
+  loop: "{{ LEO_SECRETS_PROTECTION.stdout_lines[1:] }}"
+  ignore_errors: false
+  changed_when: false
+  check_mode: false
+  when:
+    - kafka_controller_client_secrets_protection_enabled|bool
 
 - name: Remove confluent.use.controller.listener config from Client Properties
   lineinfile:
     path: "{{ kafka_controller.client_config_file }}"
     state: absent
-    line: confluent.use.controller.listener=true
+    regexp: '^confluent.use.controller.listener.*'
   changed_when: false
   check_mode: false
   when: not kraft_migration|bool
diff --git a/roles/kafka_controller/tasks/secrets_protection.yml b/roles/kafka_controller/tasks/secrets_protection.yml
index 42765bcd5..2cea2f6c5 100644
--- a/roles/kafka_controller/tasks/secrets_protection.yml
+++ b/roles/kafka_controller/tasks/secrets_protection.yml
@@ -26,6 +26,33 @@
   tags:
     - configuration
 
+- name: Add confluent.use.controller.listener config to Client Properties
+  set_fact:
+    kafka_controller_client_properties: "{{ kafka_controller_client_properties | combine(
+        {
+            'confluent.use.controller.listener': 'true'
+        }
+      ) }}"
+  tags:
+    - configuration
+
+- name: Create Kafka Controller Client Config with Secrets Protection
+  include_role:
+    name: common
+    tasks_from: secrets_protection.yml
+  vars:
+    final_properties: "{{ kafka_controller_client_properties }}"
+    encrypt_passwords: "{{ kafka_controller_client_secrets_protection_encrypt_passwords }}"
+    encrypt_properties: "{{ kafka_controller_client_secrets_protection_encrypt_properties }}"
+    config_path: "{{ kafka_controller.client_config_file}}"
+    secrets_file: "{{ kafka_controller_client_secrets_protection_file }}"
+    secrets_file_owner: "{{ kafka_controller_user }}"
+    secrets_file_group: "{{ kafka_controller_group }}"
+    ca_cert_path: "{{ kafka_controller_ca_cert_path if ssl_enabled|bool else '' }}"
+    handler: "restart Kafka Controller"
+  tags:
+    - configuration
+
 - name: Update system overrides with masterkey
   template:
     src: override.conf.j2