declare all scram users in same command

kafka-storage  seems to have a problem with already formatted dir

roles/kafka_broker/tasks/get_meta_properties.yml

@@ -1,12 +1,65 @@
 ---
+
+- name: Prepare SCRAM Users
+  set_fact:
+    scram_users_to_create: []
+
+# Only supported when kafka-controller (KRaft) is enabled
+- name: Prepare SCRAM 512 Users
+  when:
+    - "'SCRAM-SHA-512' in kafka_broker_sasl_enabled_mechanisms"
+    - kraft_combined
+  set_fact:
+    scram_users_to_create: "{{ scram_users_to_create + [ '--add-scram SCRAM-SHA-512=[name=\"'+ item.value['principal'] + '\",password=\"' + item.value['password'] + '\"]' ] }}"
+  loop: "{{ sasl_scram_users_final|dict2items }}"
+  loop_control:
+    label: "{{ item.value['principal'] }}"
+
+- name: Create SCRAM 256 Users
+  when:
+    - "'SCRAM-SHA-256' in kafka_broker_sasl_enabled_mechanisms"
+    - kraft_combined
+  set_fact:
+    scram_users_to_create: "{{ scram_users_to_create + [ '--add-scram SCRAM-SHA-256=[name=\"'+ item.value['principal'] + '\",password=\"' + item.value['password'] + '\"]' ] }}"
+  loop: "{{ sasl_scram_users_final|dict2items }}"
+  loop_control:
+    label: "{{ item.value['principal'] }}"
+
+# with kraft combined mode, first install have to define clusterid, instead of getting it from controller
+- name: Check meta.properties
+  ansible.builtin.stat:
+    path: "{{ kafka_controller_final_properties['log.dirs'] }}/meta.properties"
+  delegate_to: "{{ kafka_controller_default_host if kafka_controller_default_host != inventory_hostname else omit }}"
+  register: meta_properties
+
+- name: Initialize ClusterId
+  shell: "{{ binary_base_path }}/bin/kafka-storage random-uuid"
+  environment:
+    KAFKA_OPTS: "-Xlog:all=error -XX:+IgnoreUnrecognizedVMOptions"
+  register: random_uuid
+  run_once: true
+  when: not meta_properties.stat.exists
+
+- name: Set ClusterId
+  set_fact:
+    clusterid: "{{ random_uuid.stdout }}"
+  run_once: true
+  when: not meta_properties.stat.exists
+
+# in other cases, clusterid is still defined onto controller nodes
 - name: Extract ClusterId from meta.properties on KRaft Controller
   slurp:
     src: "{{ kafka_controller_final_properties['log.dirs'] }}/meta.properties"
-  delegate_to: "{{ groups.kafka_controller[0] }}"
+  delegate_to: "{{ kafka_controller_default_host if kafka_controller_default_host != inventory_hostname else omit }}"
   register: uuid_broker
+  when: meta_properties.stat.exists
+
+- name: Set ClusterId
+  set_fact:
+    clusterid: "{{ (uuid_broker['content'] | b64decode).partition('cluster.id=')[2].partition('\n')[0] }}"
+  run_once: true
+  when: meta_properties.stat.exists
 
 - name: Format Storage Directory
-  shell: "{{ binary_base_path }}/bin/kafka-storage  format -t {{ clusterid }} -c {{ kafka_broker.config_file }} --ignore-formatted"
+  shell: "{{ binary_base_path }}/bin/kafka-storage  format -t {{ clusterid }} -c {{ kafka_broker.config_file }} --ignore-formatted  {{ scram_users_to_create|join(' ') }}"
   register: format_meta
-  vars:
-    clusterid: "{{ (uuid_broker['content'] | b64decode).partition('cluster.id=')[2].partition('\n')[0] }}"

roles/kafka_broker/tasks/main.yml

@@ -391,7 +391,7 @@
   run_once: true
   when:
     - "'SCRAM-SHA-512' in kafka_broker_sasl_enabled_mechanisms"
-    - not kraft_enabled
+    - not kraft_enabled|bool
   no_log: "{{mask_secrets|bool}}"
 
 # Only supported when zookeeper is enabled
@@ -406,58 +406,9 @@
   run_once: true
   when:
     - "'SCRAM-SHA-256' in kafka_broker_sasl_enabled_mechanisms"
-    - not kraft_enabled
+    - not kraft_enabled|bool
   no_log: "{{ mask_sensitive_logs|bool }}"
 
-# Only supported when kafka-controller (KRaft) is enabled
-- name: Create SCRAM Users
-  run_once: true
-  delegate_to: "{{ groups.kafka_controller[0] }}"
-  when:
-    - "'SCRAM-SHA-512' in kafka_broker_sasl_enabled_mechanisms"
-    - kraft_enabled
-  block:
-    - name: Extract ClusterId from meta.properties on KRaft Controller
-      slurp:
-        src: "{{ kafka_controller_final_properties['log.dirs'] }}/meta.properties"
-      register: uuid_broker
-
-    - name: Call controller
-      shell: |
-        {{ binary_base_path }}/bin/kafka-storage format \
-          --config {{ kafka_controller.config_file }} \
-          --cluster-id {{ clusterid }} \
-          --ignore-formatted \
-          --add-scram 'SCRAM-SHA-512=[name="{{ item.value['principal'] }}",password="{{ item.value['password'] }}"]'
-      vars:
-        clusterid: "{{ (uuid_broker['content'] | b64decode).partition('cluster.id=')[2].partition('\n')[0] }}"
-      loop: "{{ sasl_scram_users_final|dict2items }}"
-      no_log: "{{mask_secrets|bool}}"
-
-# Only supported when kafka-controller (KRaft) is enabled
-- name: Create SCRAM 256 Users
-  run_once: true
-  delegate_to: "{{ groups.kafka_controller[0] }}"
-  when:
-    - "'SCRAM-SHA-256' in kafka_broker_sasl_enabled_mechanisms"
-    - kraft_enabled
-  block:
-    - name: Extract ClusterId from meta.properties on KRaft Controller
-      slurp:
-        src: "{{ kafka_controller_final_properties['log.dirs'] }}/meta.properties"
-      register: uuid_broker
-
-    - name: Call controller
-      shell: |
-        {{ binary_base_path }}/bin/kafka-storage format \
-          --config {{ kafka_controller.config_file }} \
-          --cluster-id {{ clusterid }} \
-          --ignore-formatted \
-          --add-scram 'SCRAM-SHA-256=[name="{{ item.value['principal'] }}",password="{{ item.value['password'] }}"]'
-      vars:
-        clusterid: "{{ (uuid_broker['content'] | b64decode).partition('cluster.id=')[2].partition('\n')[0] }}"
-      loop: "{{ sasl_scram_users_final|dict2items }}"
-      no_log: "{{mask_secrets|bool}}"
 
 - name: Deploy JMX Exporter Config File
   template:

roles/kafka_controller/tasks/get_meta_properties.yml

@@ -1,13 +1,63 @@
 ---
-- name: Get ClusterId
+
+- name: Prepare SCRAM Users
+  set_fact:
+    scram_users_to_create: []
+
+- name: Prepare SCRAM 512 Users
+  when:
+    - "'SCRAM-SHA-512' in kafka_broker_sasl_enabled_mechanisms"
+  set_fact:
+    scram_users_to_create: "{{ scram_users_to_create + [ '--add-scram SCRAM-SHA-512=[name=\"'+ item.value['principal'] + '\",password=\"' + item.value['password'] + '\"]' ] }}"
+  loop: "{{ sasl_scram_users_final|dict2items }}"
+  loop_control:
+    label: "{{ item.value['principal'] }}"
+
+- name: Prepare SCRAM 256 Users
+  when:
+    - "'SCRAM-SHA-256' in kafka_broker_sasl_enabled_mechanisms"
+  set_fact:
+    scram_users_to_create: "{{ scram_users_to_create + [ '--add-scram SCRAM-SHA-256=[name=\"'+ item.value['principal'] + '\",password=\"' + item.value['password'] + '\"]' ] }}"
+  loop: "{{ sasl_scram_users_final|dict2items }}"
+  loop_control:
+    label: "{{ item.value['principal'] }}"
+
+# if meta.properties does not exists , create uuid
+- name: Check meta.properties
+  ansible.builtin.stat:
+    path: "{{ kafka_controller_final_properties['log.dirs'] }}/meta.properties"
+  register: meta_properties
+
+- name: Initialize ClusterId
   shell: "{{ binary_base_path }}/bin/kafka-storage random-uuid"
   environment:
     KAFKA_OPTS: "-Xlog:all=error -XX:+IgnoreUnrecognizedVMOptions"
   register: uuid_key
   run_once: true
+  when: not meta_properties.stat.exists
+
+- name: Set ClusterId
+  set_fact:
+    clusterid: "{{ random_uuid.stdout }}"
+  run_once: true
+  when: not meta_properties.stat.exists
+
+# else, extract it from meta.properties
+- name: Extract ClusterId from meta.properties
+  slurp:
+    src: "{{ kafka_controller_final_properties['log.dirs'] }}/meta.properties"
+  register: uuid_broker
+  run_once: true
+  when: meta_properties.stat.exists
+
+- name: Set ClusterId
+  set_fact:
+    clusterid: "{{ (uuid_broker['content'] | b64decode).partition('cluster.id=')[2].partition('\n')[0] }}"
+  run_once: true
+  when: meta_properties.stat.exists
 
 - name: Format Data Directory
-  shell: "{{ binary_base_path }}/bin/kafka-storage  format -t {{ clusterid }} -c {{ kafka_controller.config_file }}  --ignore-formatted"
+  shell: "{{ binary_base_path }}/bin/kafka-storage format -t {{ clusterid }} -c {{ kafka_controller.config_file }} --ignore-formatted {{ scram_users_to_create|join(' ') }}"
   register: format_meta
   vars:
     clusterid: "{{ uuid_key.stdout }}"