Repository: confluentinc/cp-ansible
Issue: Exception SASL SCRAM User when password is vaulted

**Describe the issue**

Using your playbook with sasl_protocol to scram. If you set user password vaulted, you have an exception in this [task](https://github.com/confluentinc/cp-ansible/blob/e7523d5011dc40a62c62ab3f1dcb78132f90c0a4/roles/confluent.kafka_broker/tasks/main.yml#L124) :
fatal: [XXXXXXX]: FAILED! => {"msg": "An unhandled exception occurred while templating '{{ kafka_broker_combined_properties | combine(kafka_broker_custom_properties) }}'. [....] can only concatenate str (not \\"AnsibleVaultEncryptedUnicode\\") to str"}

I think problem raise from combine_properties function because in default configuration password is string. So if password is vault encode myabe fix it with 'toString() python method' :
``` python

    def combine_properties(self, properties_dict):
        # Loops over master properties dictionary and combines sub elements if enabled
        final_dict = {}
        for prop in properties_dict:
            if properties_dict[prop].get('enabled'):
                final_dict.update(properties_dict[prop].get('properties'))
        return final_dict
```

**To Reproduce**
All Vars Configuration :
``` yaml
    sasl_protocol: scram

    sasl_scram_users:
      admin:
        principal: "admin"
        password: !vault |
          $ANSIBLE_VAULT;1.1;AES256
          32313537636131333038303532613231653233343565303831396530393061643031303432303632
          6537306234393563343861643834336430353930643062370a663563376131653664653430363963
          62303339313937646635663032326266383439623736623462373632313934643261303634663732
          3331643634623337610a616365613231346630656564616265656231633238313932333437363461
          65626165666531656463353562323165376135653631316435653337366164653063
      client:
        principal: "client"
        password: !vault |
          $ANSIBLE_VAULT;1.1;AES256
          35653666663833386164616630353966363231633234633561663464653931393835653830303630
          3263373337396137653637383332656539373766346438370a353164383737306535646234363065
          66643863353439663064346564333062373265653036646137306531353336323931653931363032
          3231373831366531640a323161386538333966646635353462623263346234306631383635646237
          34613764663736346362616562623061333438613461616162386532373266393166

```

**Workaround**

A [workaround exist](https://github.com/ansible/ansible/issues/33067), you need to convert vault encoded string : 
``` yaml
    sasl_protocol: scram

    sasl_admin_password: !vault |
      $ANSIBLE_VAULT;1.1;AES256
      32313537636131333038303532613231653233343565303831396530393061643031303432303632
      6537306234393563343861643834336430353930643062370a663563376131653664653430363963
      62303339313937646635663032326266383439623736623462373632313934643261303634663732
      3331643634623337610a616365613231346630656564616265656231633238313932333437363461
      65626165666531656463353562323165376135653631316435653337366164653063
      
    sasl_client_password: !vault |
      $ANSIBLE_VAULT;1.1;AES256
      35653666663833386164616630353966363231633234633561663464653931393835653830303630
      3263373337396137653637383332656539373766346438370a353164383737306535646234363065
      66643863353439663064346564333062373265653036646137306531353336323931653931363032
      3231373831366531640a323161386538333966646635353462623263346234306631383635646237
      34613764663736346362616562623061333438613461616162386532373266393166

    sasl_scram_users:
      admin:
        principal: "admin"
        password: "{{sasl_admin_password|string}}"
      client:
        principal: "client"
        password: "{{sasl_client_password|string}}"
```



Changed files:
roles/confluent.variables/filter_plugins/filters.py

diff --git a/roles/confluent.variables/filter_plugins/filters.py b/roles/confluent.variables/filter_plugins/filters.py
index e8a0fc8f..429dce61 100644
--- a/roles/confluent.variables/filter_plugins/filters.py
+++ b/roles/confluent.variables/filter_plugins/filters.py
@@ -91,9 +91,15 @@ class FilterModule(object):
     def combine_properties(self, properties_dict):
         # Loops over master properties dictionary and combines sub elements if enabled
         final_dict = {}
+        properties_tmp = {}
+
         for prop in properties_dict:
             if properties_dict[prop].get('enabled'):
-                final_dict.update(properties_dict[prop].get('properties'))
+                for p in properties_dict[prop].get('properties'):
+                    properties_tmp[p] = str(properties_dict[prop].get('properties')[p])
+                final_dict.update(properties_tmp)
+                properties_tmp = {}
+        print(final_dict)
         return final_dict
 
     def split_to_dict(self, string):
@@ -121,10 +127,10 @@ class FilterModule(object):
 
             if listeners_dict[listener].get('ssl_enabled', default_ssl_enabled):
                 final_dict['listener.name.' + listener_name + '.ssl.truststore.location'] = kafka_broker_truststore_path
-                final_dict['listener.name.' + listener_name + '.ssl.truststore.password'] = kafka_broker_truststore_storepass
+                final_dict['listener.name.' + listener_name + '.ssl.truststore.password'] = str(kafka_broker_truststore_storepass)
                 final_dict['listener.name.' + listener_name + '.ssl.keystore.location'] = kafka_broker_keystore_path
-                final_dict['listener.name.' + listener_name + '.ssl.keystore.password'] = kafka_broker_keystore_storepass
-                final_dict['listener.name.' + listener_name + '.ssl.key.password'] = kafka_broker_keystore_keypass
+                final_dict['listener.name.' + listener_name + '.ssl.keystore.password'] = str(kafka_broker_keystore_storepass)
+                final_dict['listener.name.' + listener_name + '.ssl.key.password'] = str(kafka_broker_keystore_keypass)
 
             if bouncy_castle_keystore:
                 final_dict['listener.name.' + listener_name + '.ssl.keymanager.algorithm'] = 'PKIX'
@@ -147,7 +153,7 @@ class FilterModule(object):
 
             if self.normalize_sasl_protocol(listeners_dict[listener].get('sasl_protocol', default_sasl_protocol)) == 'SCRAM-SHA-512':
                 final_dict['listener.name.' + listener_name + '.sasl.enabled.mechanisms'] = 'SCRAM-SHA-512'
-                final_dict['listener.name.' + listener_name + '.scram-sha-512.sasl.jaas.config'] = 'org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"' + scram_user + '\\" password=\\"' + scram_password + '\\";'
+                final_dict['listener.name.' + listener_name + '.scram-sha-512.sasl.jaas.config'] = 'org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"' + scram_user + '\\" password=\\"' + str(scram_password) + '\\";'
 
             if self.normalize_sasl_protocol(listeners_dict[listener].get('sasl_protocol', default_sasl_protocol)) == 'SCRAM-SHA-256':
                 final_dict['listener.name.' + listener_name + '.sasl.enabled.mechanisms'] = 'SCRAM-SHA-256'
@@ -173,17 +179,17 @@ class FilterModule(object):
         }
         if listener_dict.get('ssl_enabled', default_ssl_enabled):
             final_dict[config_prefix + 'ssl.truststore.location'] = truststore_path
-            final_dict[config_prefix + 'ssl.truststore.password'] = truststore_storepass
+            final_dict[config_prefix + 'ssl.truststore.password'] = str(truststore_storepass)
 
         if listener_dict.get('ssl_mutual_auth_enabled', default_ssl_mutual_auth_enabled):
             final_dict[config_prefix + 'ssl.keystore.location'] = keystore_path
-            final_dict[config_prefix + 'ssl.keystore.password'] = keystore_storepass
-            final_dict[config_prefix + 'ssl.key.password'] = keystore_keypass
+            final_dict[config_prefix + 'ssl.keystore.password'] = str(keystore_storepass)
+            final_dict[config_prefix + 'ssl.key.password'] = str(keystore_keypass)
 
         if listener_dict.get('ssl_mutual_auth_enabled', default_ssl_mutual_auth_enabled):
             final_dict[config_prefix + 'ssl.keystore.location'] = keystore_path
-            final_dict[config_prefix + 'ssl.keystore.password'] = keystore_storepass
-            final_dict[config_prefix + 'ssl.key.password'] = keystore_keypass
+            final_dict[config_prefix + 'ssl.keystore.password'] = str(keystore_storepass)
+            final_dict[config_prefix + 'ssl.key.password'] = str(keystore_keypass)
 
         if bouncy_castle_keystore:
             final_dict[config_prefix + 'ssl.keymanager.algorithm'] = 'PKIX'
@@ -193,11 +199,11 @@ class FilterModule(object):
 
         if self.normalize_sasl_protocol(listener_dict.get('sasl_protocol', default_sasl_protocol)) == 'PLAIN' and not omit_jaas_configs:
             final_dict[config_prefix + 'sasl.mechanism'] = 'PLAIN'
-            final_dict[config_prefix + 'sasl.jaas.config'] = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=\\"' + sasl_plain_username + '\\" password=\\"' + sasl_plain_password + '\\";'
+            final_dict[config_prefix + 'sasl.jaas.config'] = 'org.apache.kafka.common.security.plain.PlainLoginModule required username=\\"' + sasl_plain_username + '\\" password=\\"' + str(sasl_plain_password) + '\\";'
 
         if self.normalize_sasl_protocol(listener_dict.get('sasl_protocol', default_sasl_protocol)) == 'SCRAM-SHA-512' and not omit_jaas_configs:
             final_dict[config_prefix + 'sasl.mechanism'] = 'SCRAM-SHA-512'
-            final_dict[config_prefix + 'sasl.jaas.config'] = 'org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"' + sasl_scram_username + '\\" password=\\"' + sasl_scram_password + '\\";'
+            final_dict[config_prefix + 'sasl.jaas.config'] = 'org.apache.kafka.common.security.scram.ScramLoginModule required username=\\"' + sasl_scram_username + '\\" password=\\"' + str(sasl_scram_password) + '\\";'
 
         if self.normalize_sasl_protocol(listener_dict.get('sasl_protocol', default_sasl_protocol)) == 'SCRAM-SHA-256' and not omit_jaas_configs:
             final_dict[config_prefix + 'sasl.mechanism'] = 'SCRAM-SHA-256'
@@ -216,7 +222,7 @@ class FilterModule(object):
                 final_dict[config_prefix + 'sasl.login.callback.handler.class'] = 'io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler'
 
             if self.normalize_sasl_protocol(listener_dict.get('sasl_protocol', default_sasl_protocol)) == 'OAUTHBEARER' and not omit_jaas_configs:
-                final_dict[config_prefix + 'sasl.jaas.config'] = 'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required username=\\"' + oauth_username + '\\" password=\\"' + oauth_password + '\\" metadataServerUrls=\\"' + mds_bootstrap_server_urls + '\\";'
+                final_dict[config_prefix + 'sasl.jaas.config'] = 'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required username=\\"' + oauth_username + '\\" password=\\"' + str(oauth_password) + '\\" metadataServerUrls=\\"' + mds_bootstrap_server_urls + '\\";'
 
         return final_dict
 
@@ -241,10 +247,10 @@ class FilterModule(object):
 
                 if hostvars[groups[ansible_group][0]].get('kafka_connect_ssl_enabled', ssl_enabled):
                     final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.truststore.location'] = truststore_path
-                    final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.truststore.password'] = truststore_storepass
+                    final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.truststore.password'] = str(truststore_storepass)
                     final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.keystore.location'] = keystore_path
-                    final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.keystore.password'] = keystore_storepass
-                    final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.key.password'] = keystore_keypass
+                    final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.keystore.password'] = str(keystore_storepass)
+                    final_dict['confluent.controlcenter.connect.' + hostvars[groups[ansible_group][0]].get('kafka_connect_group_id', default_connect_group_id) + '.ssl.key.password'] = str(keystore_keypass)
 
         return final_dict
 
@@ -272,9 +278,9 @@ class FilterModule(object):
 
                 if hostvars[groups[ansible_group][0]].get('ksql_ssl_enabled', ssl_enabled):
                     final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.truststore.location'] = truststore_path
-                    final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.truststore.password'] = truststore_storepass
+                    final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.truststore.password'] = str(truststore_storepass)
                     final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.keystore.location'] = keystore_path
-                    final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.keystore.password'] = keystore_storepass
-                    final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.key.password'] = keystore_keypass
+                    final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.keystore.password'] = str(keystore_storepass)
+                    final_dict['confluent.controlcenter.ksql.' + ansible_group + '.ssl.key.password'] = str(keystore_keypass)
 
         return final_dict