main.yml

---
confluent_ansible_branch: 8.2.0-post

systemd_base_dir: "{{'/lib/systemd/system' if ansible_os_family == 'Debian' else '/usr/lib/systemd/system'}}"

confluent_repo_version: "{{ confluent_package_version | regex_replace('^([0-9])\\.([0-9]*).*', '\\1.\\2') }}"

# Confirm no trailing / on the ssl file directory path
ssl_file_dir_final: "{{ ssl_file_dir |regex_replace('\\/$', '') }}"

### The base path for the binary files. When in Archive File deployment mode this results in binary files being based in, for example `/opt/confluent/confluent-5.5.1/bin`, otherwise they are based in `/usr/bin`.
base_path: "{{ ((config_base_path,('confluent-',archive_version) | join) | path_join) if installation_method == 'archive' else '/' }}"

binary_base_path: "{{ ((config_base_path,('confluent-',archive_version) | join) | path_join) if installation_method == 'archive' else '/usr' }}"

confluent_control_center_next_gen_base_path: "{{ ((config_base_path,('confluent-control-center-next-gen-',confluent_control_center_next_gen_archive_version) | join) | path_join) if installation_method == 'archive' else '/' }}"

confluent_control_center_next_gen_binary_base_path: "{{ ((config_base_path,('confluent-control-center-next-gen-',confluent_control_center_next_gen_archive_version) | join) | path_join) if installation_method == 'archive' else '/usr' }}"

### Runs kafka in Kraft mode 8.0.x onwards
kraft_enabled: true

#### Config prefix paths ####
kafka_controller_config_prefix_path: "{{ kafka_controller_config_prefix.strip('/') }}"
kafka_broker_config_prefix_path: "{{ kafka_broker_config_prefix.strip('/') }}"
schema_registry_config_prefix_path: "{{ schema_registry_config_prefix.strip('/') }}"
kafka_rest_config_prefix_path: "{{ kafka_rest_config_prefix.strip('/') }}"
kafka_connect_config_prefix_path: "{{ kafka_connect_config_prefix.strip('/') }}"
ksql_config_prefix_path: "{{ ksql_config_prefix.strip('/') }}"
control_center_next_gen_config_prefix_path: "{{ control_center_next_gen_config_prefix.strip('/') }}"
kafka_connect_replicator_config_prefix_path: "{{ kafka_connect_replicator_config_prefix.strip('/') }}"

# Jolokia vars
kafka_broker_jolokia_java_arg_ssl_addon: ",keystore={{kafka_broker_keystore_path}},keystorePassword={{kafka_broker_keystore_storepass}},protocol=https"
kafka_broker_jolokia_urp_url: "{{ 'https' if kafka_broker_jolokia_ssl_enabled|bool else 'http' }}://{{ hostvars[inventory_hostname] | confluent.platform.resolve_and_format_hostname }}:{{kafka_broker_jolokia_port}}/jolokia/read/kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"
kafka_broker_jolokia_active_controller_url: "{{ 'https' if kafka_broker_jolokia_ssl_enabled|bool else 'http' }}://{{ hostvars[inventory_hostname] | confluent.platform.resolve_and_format_hostname }}:{{kafka_broker_jolokia_port}}/jolokia/read/kafka.controller:type=KafkaController,name=ActiveControllerCount"

# OAuth vars
oauth_enabled: "{{ 'oauth' in auth_mode }}"
ldap_with_oauth_enabled: "{{ 'ldap_with_oauth' in auth_mode }}"
schema_registry_oauth_enabled: "{{ 'oauth' in schema_registry_auth_mode }}"
kafka_rest_oauth_enabled: "{{ 'oauth' in kafka_rest_auth_mode }}"
kafka_connect_oauth_enabled: "{{ 'oauth' in kafka_connect_auth_mode }}"
kafka_connect_replicator_oauth_enabled: "{{ 'oauth' in kafka_connect_replicator_auth_mode }}"
ksql_oauth_enabled: "{{ 'oauth' in ksql_auth_mode }}"

# oauth client assertion vars
oauth_superuser_third_party_client_assertion_config:
  superuser: "{{ (oauth_superuser_client_assertion_file_base_path != 'none') | ternary(oauth_superuser_client_assertion_file_base_path ~ '/superuser_client.jwt', 'none') }}"

ksql_oauth_client_assertion_config:
  enabled: "{{ksql_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{ksql_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{ksql_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{ksql_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ ksql_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ ksql_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{ksql_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{ksql_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{ksql_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{ksql_oauth_client_assertion_file_base_path}}"

ksql_third_party_oauth_client_assertion_config:
  kafka: "{{ (ksql_oauth_client_assertion_file_base_path != 'none') | ternary(ksql_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"
  schema_registry: "{{ (ksql_oauth_client_assertion_file_base_path != 'none') | ternary(ksql_oauth_client_assertion_file_base_path ~ '/schema_registry_client.jwt', 'none') }}"
  mds: "{{ (ksql_oauth_client_assertion_file_base_path != 'none') | ternary(ksql_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  monitoring_interceptor: "{{ (ksql_oauth_client_assertion_file_base_path != 'none') | ternary(ksql_oauth_client_assertion_file_base_path ~ '/monitoring_interceptor_client.jwt', 'none') }}"
  ksql: "{{ (ksql_oauth_client_assertion_file_base_path != 'none') | ternary(ksql_oauth_client_assertion_file_base_path ~ '/ksql_client.jwt', 'none') }}"

kafka_rest_oauth_client_assertion_config:
  enabled: "{{kafka_rest_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_rest_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_rest_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_rest_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_rest_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_rest_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_rest_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_rest_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_rest_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_rest_oauth_client_assertion_file_base_path}}"

kafka_rest_third_party_oauth_client_assertion_config:
  license: "{{ (kafka_rest_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_rest_oauth_client_assertion_file_base_path ~ '/license_client.jwt', 'none') }}"
  kafka: "{{ (kafka_rest_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_rest_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"
  monitoring_interceptor: "{{ (kafka_rest_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_rest_oauth_client_assertion_file_base_path ~ '/monitoring_interceptor_client.jwt', 'none') }}"
  mds: "{{ (kafka_rest_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_rest_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  kafka_rest: "{{ (kafka_rest_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_rest_oauth_client_assertion_file_base_path ~ '/kafka_rest_client.jwt', 'none') }}"

kafka_connect_oauth_client_assertion_config:
  enabled: "{{kafka_connect_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_connect_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_connect_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_connect_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_connect_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_connect_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_connect_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_connect_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_connect_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_connect_oauth_client_assertion_file_base_path}}"

kafka_connect_third_party_oauth_client_assertion_config:
  kafka: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"
  producer: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/producer_client.jwt', 'none') }}"
  consumer: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/consumer_client.jwt', 'none') }}"
  mds: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  producer_monitoring_interceptor: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/producer_monitoring_interceptor_client.jwt', 'none') }}"
  consumer_monitoring_interceptor: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/consumer_monitoring_interceptor_client.jwt', 'none') }}"
  secret_registry: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/secret_registry_client.jwt', 'none') }}"
  kafka_connect: "{{ (kafka_connect_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_oauth_client_assertion_file_base_path ~ '/kafka_connect_client.jwt', 'none') }}"

schema_registry_oauth_client_assertion_config:
  enabled: "{{schema_registry_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{schema_registry_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{schema_registry_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{schema_registry_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ schema_registry_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ schema_registry_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{schema_registry_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{ schema_registry_oauth_client_assertion_jti_include|bool }}"
  client_assertion_nbf_include: "{{schema_registry_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{schema_registry_oauth_client_assertion_file_base_path}}"

schema_registry_third_party_oauth_client_assertion_config:
  kafka: "{{ (schema_registry_oauth_client_assertion_file_base_path != 'none') | ternary(schema_registry_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"
  mds: "{{ (schema_registry_oauth_client_assertion_file_base_path != 'none') | ternary(schema_registry_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  schema_registry: "{{ (schema_registry_oauth_client_assertion_file_base_path != 'none') | ternary(schema_registry_oauth_client_assertion_file_base_path ~ '/schema_registry_client.jwt', 'none') }}"

kafka_controller_oauth_client_assertion_config:
  enabled: "{{kafka_controller_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_controller_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_controller_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_controller_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{kafka_controller_oauth_client_assertion_private_key_file_dest_path}}"
  client_assertion_template_file: "{{ kafka_controller_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_controller_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_controller_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_controller_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_controller_oauth_client_assertion_file_base_path}}"

kafka_controller_third_party_oauth_client_assertion_config:
  mds: "{{ (kafka_controller_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_controller_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  metric_reporters: "{{ (kafka_controller_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_controller_oauth_client_assertion_file_base_path ~ '/metric_reporters_client.jwt', 'none') }}"
  audit_logs_destination_client: "{{ (kafka_controller_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_controller_oauth_client_assertion_file_base_path ~ '/audit_logs_destination_client.jwt', 'none') }}"
  audit_logs_destination_admin_client: "{{ (kafka_controller_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_controller_oauth_client_assertion_file_base_path ~ '/audit_logs_destination_admin_client.jwt', 'none') }}"
  controller: "{{ (kafka_controller_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_controller_oauth_client_assertion_file_base_path ~ '/controller_client.jwt', 'none') }}"

kafka_broker_oauth_client_assertion_config:
  enabled: "{{kafka_broker_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_broker_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_broker_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_broker_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_broker_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_broker_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_broker_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_broker_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_broker_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_broker_oauth_client_assertion_file_base_path}}"

kafka_broker_third_party_oauth_client_assertion_config:
  mds: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  embedded_rest_proxy: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/embedded_rest_proxy_client.jwt', 'none') }}"
  embedded_rest_proxy_rbac_oauth: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/embedded_rest_proxy_rbac_oauth_client.jwt', 'none') }}"
  metrics_reporter: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/metrics_reporter_client.jwt', 'none') }}"
  audit_logs_destination_client: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/audit_logs_destination_client.jwt', 'none') }}"
  audit_logs_destination_admin_client: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/audit_logs_destination_admin_client.jwt', 'none') }}"
  schema_registry: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/schema_registry_client.jwt', 'none') }}"
  kafka_broker: "{{ (kafka_broker_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_broker_oauth_client_assertion_file_base_path ~ '/kafka_broker_client.jwt', 'none') }}"

kafka_connect_replicator_oauth_client_assertion_config:
  enabled: "{{kafka_connect_replicator_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_connect_replicator_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_connect_replicator_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_connect_replicator_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_connect_replicator_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_connect_replicator_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_connect_replicator_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_connect_replicator_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_connect_replicator_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_connect_replicator_oauth_client_assertion_file_base_path}}"

kafka_connect_replicator_third_party_oauth_client_assertion_config:
  kafka: "{{ (kafka_connect_replicator_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"
  mds: "{{ (kafka_connect_replicator_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_oauth_client_assertion_file_base_path ~ '/mds_client.jwt', 'none') }}"
  kafka_connect_replicator: "{{ (kafka_connect_replicator_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_oauth_client_assertion_file_base_path ~ '/kafka_connect_replicator_client.jwt', 'none') }}"

kafka_connect_replicator_producer_oauth_client_assertion_config:
  enabled: "{{kafka_connect_replicator_producer_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_connect_replicator_producer_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_connect_replicator_producer_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_connect_replicator_producer_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_connect_replicator_producer_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_connect_replicator_producer_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_connect_replicator_producer_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_connect_replicator_producer_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_connect_replicator_producer_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_connect_replicator_producer_oauth_client_assertion_file_base_path}}"

kafka_connect_replicator_producer_third_party_oauth_client_assertion_config:
  kafka: "{{ (kafka_connect_replicator_producer_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_producer_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"

kafka_connect_replicator_consumer_oauth_client_assertion_config:
  enabled: "{{kafka_connect_replicator_consumer_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_connect_replicator_consumer_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_connect_replicator_consumer_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_connect_replicator_consumer_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_connect_replicator_consumer_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_connect_replicator_consumer_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_connect_replicator_consumer_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_connect_replicator_consumer_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_connect_replicator_consumer_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_connect_replicator_consumer_oauth_client_assertion_file_base_path}}"

kafka_connect_replicator_consumer_third_party_oauth_client_assertion_config:
  kafka: "{{ (kafka_connect_replicator_consumer_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_consumer_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"

kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_config:
  enabled: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_file_base_path}}"

kafka_connect_replicator_monitoring_interceptor_third_party_oauth_client_assertion_config:
  kafka: "{{ (kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"

control_center_next_gen_oauth_client_assertion_config:
  enabled: "{{control_center_next_gen_oauth_client_assertion_enabled}}"
  client_assertion_issuer: "{{control_center_next_gen_oauth_client_assertion_issuer}}"
  client_assertion_sub: "{{control_center_next_gen_oauth_client_assertion_sub}}"
  client_assertion_audience: "{{control_center_next_gen_oauth_client_assertion_audience}}"
  client_assertion_private_key_file: "{{ control_center_next_gen_oauth_client_assertion_private_key_file_dest_path }}"
  client_assertion_template_file: "{{ control_center_next_gen_oauth_client_assertion_template_file_dest_path }}"
  client_assertion_private_key_passphrase: "{{control_center_next_gen_oauth_client_assertion_private_key_passphrase}}"
  client_assertion_jti_include: "{{control_center_next_gen_oauth_client_assertion_jti_include}}"
  client_assertion_nbf_include: "{{control_center_next_gen_oauth_client_assertion_nbf_include}}"
  client_assertion_file: "{{control_center_next_gen_oauth_client_assertion_file_base_path}}"

control_center_next_gen_third_party_oauth_client_assertion_config:
  kafka: "{{ (control_center_next_gen_oauth_client_assertion_file_base_path != 'none') | ternary(control_center_next_gen_oauth_client_assertion_file_base_path ~ '/kafka_client.jwt', 'none') }}"

kafka_connect_replicator_erp_third_party_oauth_client_assertion_config:
  erp: "{{ (kafka_connect_replicator_erp_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_erp_oauth_client_assertion_file_base_path ~ '/erp.jwt', 'none') }}"

kafka_connect_replicator_producer_erp_third_party_oauth_client_assertion_config:
  producer_erp: "{{ (kafka_connect_replicator_erp_producer_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_erp_producer_oauth_client_assertion_file_base_path ~ '/producer_erp.jwt', 'none') }}"

kafka_connect_replicator_consumer_erp_third_party_oauth_client_assertion_config:
  consumer_erp: "{{ (kafka_connect_replicator_erp_consumer_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_erp_consumer_oauth_client_assertion_file_base_path ~ '/consumer_erp.jwt', 'none') }}"

kafka_connect_replicator_monitoring_interceptor_erp_third_party_oauth_client_assertion_config:
  interceptor_erp: "{{ (kafka_connect_replicator_monitoring_interceptor_erp_third_party_oauth_client_assertion_file_base_path != 'none') | ternary(kafka_connect_replicator_monitoring_interceptor_erp_third_party_oauth_client_assertion_file_base_path ~ '/erp_interceptor.jwt', 'none') }}"

#### Kafka controller variables
kafka_controller_service_name: "confluent-kcontroller"
kafka_controller_main_package: "{{ 'confluent-server' if confluent_server_enabled|bool else 'confluent-kafka'}}"
kafka_controller_default_group: confluent
kafka_controller_default_user: cp-kafka
kafka_controller_default_log_dir: /var/log/controller
kafka_controller:
  server_start_file: "{{ binary_base_path }}/bin/kafka-server-start"
  config_file: "{{ (config_base_path, kafka_controller_config_prefix_path, 'server.properties') | path_join }}"
  client_config_file: "{{ (config_base_path, kafka_controller_config_prefix_path, 'client.properties') | path_join }}"
  systemd_file: "{{systemd_base_dir}}/{{kafka_controller_service_name}}.service"
  systemd_override: /etc/systemd/system/{{kafka_controller_service_name}}.service.d/override.conf
  log4j_file: "{{ (base_path, 'etc/kafka/log4j2.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, kafka_controller_config_prefix_path, 'kafka_server_jaas.conf') | path_join }}"

kafka_controller_properties:
  defaults:
    enabled: true
    properties:
      group.initial.rebalance.delay.ms: 3000
      log.retention.check.interval.ms: 300000
      log.retention.hours: 168
      log.segment.bytes: 1073741824
      num.io.threads: 16
      num.network.threads: 8
      num.partitions: 1
      num.recovery.threads.per.data.dir: 2
      offsets.topic.replication.factor: "{{kafka_controller_default_internal_replication_factor}}"
      socket.receive.buffer.bytes: 102400
      socket.request.max.bytes: 104857600
      socket.send.buffer.bytes: 102400
      transaction.state.log.min.isr: "{{ [ 2, kafka_controller_default_internal_replication_factor|int ] | min }}"
      transaction.state.log.replication.factor: "{{kafka_controller_default_internal_replication_factor}}"
      confluent.ansible.managed: 'true'
      confluent.license.topic: _confluent-command
      confluent.license.topic.replication.factor: "{{kafka_controller_default_internal_replication_factor}}"
      confluent.metadata.topic.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      confluent.balancer.topic.replication.factor: "{{kafka_controller_default_internal_replication_factor}}"
      confluent.security.event.logger.exporter.kafka.topic.replicas: "{{audit_logs_destination_bootstrap_servers.split(',')|length if audit_logs_destination_enabled and rbac_enabled else kafka_controller_default_internal_replication_factor}}"
      confluent.support.metrics.enable: "true"
      confluent.support.customer.id: anonymous
      log.dirs: "/var/lib/controller/data"
      kafka.rest.enable: "{{kafka_controller_rest_proxy_enabled|string|lower}}"
      process.roles: controller
      controller.quorum.voters: "{{ kafka_controller_quorum_voters }}"
      controller.listener.names: "{{kafka_controller_listeners['controller']['name']}}"
      listener.security.protocol.map: "{% for listener in kafka_controller_listeners|dict2items %}{% if loop.index > 1%},{% endif %}{{ listener['value']['name'] }}:{{ listener['value'] | confluent.platform.kafka_protocol_defaults(kafka_controller_ssl_enabled, kafka_controller_sasl_protocol)}}{% endfor %},{{kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['name']}}:{{kafka_broker_listeners[kafka_broker_inter_broker_listener_name] | confluent.platform.kafka_protocol_defaults(ssl_enabled, sasl_protocol)}}"
      listeners: "{{kafka_controller_listeners['controller']['name']}}://:{{kafka_controller_listeners['controller']['port']}}"
      advertised.listeners: "{{kafka_controller_listeners['controller']['name']}}://{{ kafka_controller_listeners['controller']['hostname'] | default(hostvars[inventory_hostname]|confluent.platform.resolve_hostname) | confluent.platform.format_hostname }}:{{kafka_controller_listeners['controller']['port']}}"
      node.id: "{{ node_id|default(groups.kafka_controller.index(inventory_hostname)|int + 9991) if inventory_hostname in groups.kafka_controller else 0}}"
      inter.broker.listener.name: "{{kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['name']}}"
  broker_listener:
    enabled: true
    properties: "{{  {'broker_listener': kafka_broker_listeners[kafka_broker_inter_broker_listener_name]} | confluent.platform.listener_properties(ssl_enabled, fips_enabled, ssl_client_authentication, principal_mapping_rules, sasl_protocol,
                    kafka_controller_truststore_path, kafka_controller_truststore_storepass, kafka_controller_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                    plain_jaas_config, kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'), kerberos_kafka_controller_primary,
                    sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password, rbac_enabled_public_pem_path, oauth_enabled, oauth_jwks_uri, oauth_expected_audience, oauth_sub_claim, rbac_enabled, false, false) }}"
  kraft_sasl_plain:
    enabled: "{{(kafka_controller_sasl_protocol | confluent.platform.split_to_list)[0] == 'plain'}}"
    properties:
      sasl.mechanism.controller.protocol: PLAIN
  kraft_sasl_gssapi:
    enabled: "{{(kafka_controller_sasl_protocol | confluent.platform.split_to_list)[0] == 'kerberos'}}"
    properties:
      sasl.mechanism.controller.protocol: GSSAPI
  ssl:
    enabled: "{{ kafka_controller_listeners | confluent.platform.ssl_required(kafka_controller_ssl_enabled) }}"
    properties:
      ssl.truststore.location: "{{kafka_controller_truststore_path}}"
      ssl.truststore.password: "{{kafka_controller_truststore_storepass}}"
      ssl.keystore.location: "{{kafka_controller_keystore_path}}"
      ssl.keystore.password: "{{kafka_controller_keystore_storepass}}"
      ssl.key.password: "{{kafka_controller_keystore_storepass}}"
  principal_mapping_rules:
    enabled: "{{ssl_enabled}}"
    properties:
      ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  fips:
    enabled: "{{fips_enabled}}"
    properties:
      enable.fips: 'true'
      enable.fips.mode: "{{fips_mode}}"
      security.providers: io.confluent.kafka.security.fips.provider.BcFipsProviderCreator,io.confluent.kafka.security.fips.provider.BcFipsJsseProviderCreator
      ssl.keymanager.algorithm: PKIX
      ssl.trustmanager.algorithm: PKIX
      ssl.keystore.type: BCFKS
      ssl.truststore.type: BCFKS
      ssl.truststore.location: "{{kafka_controller_truststore_path}}"
      ssl.truststore.password: "{{kafka_controller_truststore_storepass}}"
      ssl.keystore.location: "{{kafka_controller_keystore_path}}"
      ssl.keystore.password: "{{kafka_controller_keystore_storepass}}"
      ssl.key.password: "{{kafka_controller_keystore_storepass}}"
      ssl.enabled.protocols: TLSv1.2,TLSv1.3
  sasl_enabled:
    enabled: "{{ kafka_sasl_enabled_mechanisms|length > 0 }}"
    properties:
      sasl.enabled.mechanisms: "{% for mechanism in kafka_sasl_enabled_mechanisms %}{% if loop.index > 1%},{% endif %}{{ mechanism|upper }}{% endfor %}"
  sasl_gssapi:
    enabled: "{{ 'GSSAPI' in kafka_sasl_enabled_mechanisms }}"
    properties:
      sasl.kerberos.service.name: "{{kerberos_kafka_controller_primary}}"
  inter_broker_sasl:
    enabled: "{{ 'none' not in (kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol) }}"
    properties:
      sasl.mechanism.inter.broker.protocol: "{{ (kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] }}"
  sr:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups }}"
    properties:
      confluent.schema.registry.url: "{{schema_registry_url}}"
  rbac:
    enabled: "{{ rbac_enabled }}"
    properties:
      authorizer.class.name: io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
      confluent.authorizer.access.rule.providers: CONFLUENT,KRAFT_ACL
      super.users: >-
        {{
           ('ldap' in auth_mode) | ternary("User:" + mds_super_user|default('mds'), "") +
           (auth_mode == 'ldap_with_oauth') | ternary(";", "") +
           ('oauth' in auth_mode) | ternary("User:" + oauth_superuser_principal, "")
        }}
  rbac_mds:
    enabled: "{{rbac_enabled and not external_mds_enabled}}"
    properties:
      confluent.metadata.server.kraft.controller.enabled: true
      confluent.metadata.bootstrap.servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['port']}}"
  rbac_external_mds:
    enabled: "{{rbac_enabled and external_mds_enabled}}"
    properties:
      confluent.metadata.bootstrap.servers: "{{mds_broker_bootstrap_servers}}"
  rbac_external_mds_client:
    enabled: "{{rbac_enabled}}"
    properties: "{{ mds_broker_listener | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                            'confluent.metadata.', kafka_controller_truststore_path, kafka_controller_truststore_storepass, False, kafka_controller_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                            false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                            kerberos_kafka_controller_primary, kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'),
                            false, kafka_controller_ldap_user, kafka_controller_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_controller_oauth_user, kafka_controller_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_controller_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_controller_third_party_oauth_client_assertion_config.mds)) }}"
  rbac_external_mds_oauth_client:
    enabled: "{{rbac_enabled and oauth_enabled and idp_self_signed}}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{kafka_controller_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_controller_truststore_storepass}}"
  oauth_claim:
    enabled: "{{oauth_enabled and oauth_groups_claim != 'none'}}"
    properties:
      confluent.oauth.groups.claim.name: "{{oauth_groups_claim}}"
  listeners:
    enabled: true
    properties: "{{ kafka_controller_listeners | confluent.platform.listener_properties(kafka_controller_ssl_enabled, fips_enabled, kafka_controller_ssl_client_authentication, principal_mapping_rules, kafka_controller_sasl_protocol,
                    kafka_controller_truststore_path, kafka_controller_truststore_storepass, kafka_controller_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                    plain_jaas_config, kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'), kerberos_kafka_controller_primary,
                    sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password, rbac_enabled_public_pem_path, oauth_enabled, oauth_jwks_uri, oauth_expected_audience, oauth_sub_claim, rbac_enabled, true, false) }}"
  metrics_reporter_for_control_center_next_gen:
    enabled: "{{ kafka_controller_metrics_reporter_for_control_center_next_gen_enabled|bool }}"
    properties:
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter,org.apache.kafka.common.metrics.JmxReporter
      confluent.telemetry.exporter._c3.type: "http"
      confluent.telemetry.exporter._c3.enabled: "true"
      confluent.telemetry.exporter._c3.metrics.include: "io.confluent.kafka.server.request.(?!.*delta).*|io.confluent.kafka.server.server.broker.state|io.confluent.kafka.server.replica.manager.leader.count|io.confluent.kafka.server.request.queue.size|io.confluent.kafka.server.broker.topic.failed.produce.requests.rate.1.min|io.confluent.kafka.server.tier.archiver.total.lag|io.confluent.kafka.server.request.total.time.ms.p99|io.confluent.kafka.server.broker.topic.failed.fetch.requests.rate.1.min|io.confluent.kafka.server.broker.topic.total.fetch.requests.rate.1.min|io.confluent.kafka.server.partition.caught.up.replicas.count|io.confluent.kafka.server.partition.observer.replicas.count|io.confluent.kafka.server.tier.tasks.num.partitions.in.error|io.confluent.kafka.server.broker.topic.bytes.out.rate.1.min|io.confluent.kafka.server.request.total.time.ms.p95|io.confluent.kafka.server.controller.active.controller.count|io.confluent.kafka.server.session.expire.listener.zookeeper.disconnects.total|io.confluent.kafka.server.request.total.time.ms.p999|io.confluent.kafka.server.controller.active.broker.count|io.confluent.kafka.server.request.handler.pool.request.handler.avg.idle.percent.rate.1.min|io.confluent.kafka.server.session.expire.listener.zookeeper.disconnects.rate.1.min|io.confluent.kafka.server.controller.unclean.leader.elections.rate.1.min|io.confluent.kafka.server.replica.manager.partition.count|io.confluent.kafka.server.controller.unclean.leader.elections.total|io.confluent.kafka.server.partition.replicas.count|io.confluent.kafka.server.broker.topic.total.produce.requests.rate.1.min|io.confluent.kafka.server.controller.offline.partitions.count|io.confluent.kafka.server.socket.server.network.processor.avg.idle.percent|io.confluent.kafka.server.partition.under.replicated|io.confluent.kafka.server.log.log.start.offset|io.confluent.kafka.server.log.tier.size|io.confluent.kafka.server.log.size|io.confluent.kafka.server.tier.fetcher.bytes.fetched.total|io.confluent.kafka.server.request.total.time.ms.p50|io.confluent.kafka.server.tenant.consumer.lag.offsets|io.confluent.kafka.server.session.expire.listener.zookeeper.expires.rate.1.min|io.confluent.kafka.server.log.log.end.offset|io.confluent.kafka.server.broker.topic.bytes.in.rate.1.min|io.confluent.kafka.server.partition.under.min.isr|io.confluent.kafka.server.partition.in.sync.replicas.count|io.confluent.telemetry.http.exporter.batches.dropped|io.confluent.telemetry.http.exporter.items.total|io.confluent.telemetry.http.exporter.items.succeeded|io.confluent.telemetry.http.exporter.send.time.total.millis|io.confluent.kafka.server.controller.leader.election.rate.(?!.*delta).*|io.confluent.telemetry.http.exporter.batches.failed|io.confluent.kafka.server.socket_server.connections|org.apache.kafka.consumer.(fetch.manager.fetch.latency.avg|connection.creation.total|fetch.manager.fetch.total|fetch.manager.bytes.consumed.rate)|org.apache.kafka.producer.(bufferpool.wait.ratio|request.total|topic.byte.rate)|io.confluent.kafka.server.controller.fenced.broker.count"
      confluent.telemetry.exporter._c3.client.base.url: "{{control_center_next_gen_dependency_prometheus_url}}/api/v1/otlp"
      confluent.telemetry.exporter._c3.client.compression: gzip
      confluent.telemetry.exporter._c3.api.key: "{{kafka_controller_telemetry_control_center_next_gen_user}}"
      confluent.telemetry.exporter._c3.api.secret: "{{kafka_controller_telemetry_control_center_next_gen_password}}"
      confluent.telemetry.exporter._c3.buffer.pending.batches.max: 80
      confluent.telemetry.exporter._c3.buffer.batch.items.max: 4000
      confluent.telemetry.exporter._c3.buffer.inflight.submissions.max: 10
      confluent.telemetry.metrics.collector.interval.ms: 60000
      confluent.telemetry.remoteconfig._confluent.enabled: false
      confluent.consumer.lag.emitter.enabled: "true"
  metrics_reporter_for_control_center_next_gen_client:
    enabled: "{{control_center_next_gen_dependency_prometheus_ssl_enabled}}"
    properties:
      confluent.telemetry.exporter._c3.https.ssl.protocol: TLSv1.2
      confluent.telemetry.exporter._c3.https.ssl.truststore.location: "{{kafka_controller_truststore_path}}"
      confluent.telemetry.exporter._c3.https.ssl.truststore.password: "{{kafka_controller_truststore_storepass}}"
      # If FIPS set to BCFKS else JKS
      confluent.telemetry.exporter._c3.https.ssl.truststore.type: "{% if fips_enabled|bool %}BCFKS{% else %}JKS{% endif %}"
  metrics_reporter_for_control_center_next_gen_client_mtls:
    enabled: "{{control_center_next_gen_dependency_prometheus_mtls_enabled}}"
    properties:
      confluent.telemetry.exporter._c3.https.ssl.keystore.location: "{{kafka_controller_keystore_path}}"
      confluent.telemetry.exporter._c3.https.ssl.keystore.password: "{{kafka_controller_keystore_storepass}}"
      confluent.telemetry.exporter._c3.https.ssl.key.password: "{{kafka_controller_keystore_keypass}}"
      # If FIPS set to BCFKS else JKS
      confluent.telemetry.exporter._c3.https.ssl.keystore.type: "{% if fips_enabled|bool %}BCFKS{% else %}JKS{% endif %}"
  metrics_reporter_client:
    enabled: "{{ kafka_controller_metrics_reporter_enabled|bool }}"
    properties: "{{ kafka_broker_listeners[kafka_broker_inter_broker_listener_name] | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.metrics.reporter.', kafka_controller_truststore_path, kafka_controller_truststore_storepass, False, kafka_controller_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_controller_primary, kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'),
                    false, kafka_controller_ldap_user, kafka_controller_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_controller_oauth_user, kafka_controller_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_controller_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_controller_third_party_oauth_client_assertion_config.metric_reporters)) }}"
  telemetry:
    enabled: "{{kafka_controller_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
  telemetry_proxy:
    enabled: "{{kafka_controller_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{kafka_controller_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{kafka_controller_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"
  audit_logs_destination:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled}}"
    properties:
      confluent.security.event.logger.exporter.kafka.bootstrap.servers: "{{audit_logs_destination_bootstrap_servers}}"
      confluent.security.event.logger.exporter.kafka.topic.create: 'false'
  audit_logs_destination_client:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled}}"
    properties: "{{ audit_logs_destination_listener | confluent.platform.client_properties(kafka_controller_ssl_enabled, fips_enabled, kafka_controller_ssl_mutual_auth_enabled, kafka_controller_sasl_protocol,
                    'confluent.security.event.logger.exporter.kafka.', kafka_controller_truststore_path, kafka_controller_truststore_storepass, False, kafka_controller_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_controller_primary|default('kafka'), kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'),
                    false, 'user', 'pass', mds_bootstrap_server_urls, oauth_enabled, kafka_controller_oauth_user, kafka_controller_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, true, kafka_controller_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_controller_third_party_oauth_client_assertion_config.audit_logs_destination_client)) }}"
  audit_logs_destination_admin:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled and not external_mds_enabled}}"
    properties:
      confluent.security.event.logger.destination.admin.bootstrap.servers: "{{audit_logs_destination_bootstrap_servers}}"
  audit_logs_destination_admin_client:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled and not external_mds_enabled}}"
    properties: "{{ audit_logs_destination_listener | confluent.platform.client_properties(kafka_controller_ssl_enabled, fips_enabled, kafka_controller_ssl_mutual_auth_enabled, kafka_controller_sasl_protocol,
                    'confluent.security.event.logger.destination.admin.', kafka_controller_truststore_path, kafka_controller_truststore_storepass, False, kafka_controller_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_controller_primary|default('kafka'), kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'),
                    false, 'user', 'pass', mds_bootstrap_server_urls, oauth_enabled, kafka_controller_oauth_user, kafka_controller_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, true, kafka_controller_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_controller_third_party_oauth_client_assertion_config.audit_logs_destination_admin_client)) }}"
  usm_agent_telemetry:
    enabled: "{{ 'usm_agent' in groups }}"
    properties:
      confluent.telemetry.exporter._usm.enabled: true
      confluent.telemetry.exporter._usm.events.enabled: true
      confluent.telemetry.exporter._usm.type: http
      confluent.telemetry.exporter._usm.buffer.batch.items.max: 4000
      confluent.telemetry.exporter._usm.buffer.inflight.submissions.max: 10
      confluent.telemetry.exporter._usm.buffer.pending.batches.max: 80
      confluent.telemetry.exporter._usm.client.base.url: "{{ usm_agent_url }}"
      confluent.telemetry.exporter._usm.events.client.base.url: "{{ usm_agent_url }}"
      confluent.catalog.collector.enable: true
      confluent.catalog.collector.destination.topic: catalog-events
      confluent.catalog.collector.full.configs.enable: true
      confluent.catalog.collector.multitenant.topics.enable: false
      confluent.telemetry.exporter._usm.events.filtering.routes.allowed: catalog-events
      confluent.telemetry.exporter._usm.events.filtering.enabled: true
  usm_agent_telemetry_auth_basic:
    enabled: "{{ 'usm_agent' in groups and usm_agent_basic_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.api.key: "{{ kafka_controller_usm_agent_username }}"
      confluent.telemetry.exporter._usm.api.secret: "{{ kafka_controller_usm_agent_password }}"
      confluent.telemetry.exporter._usm.events.api.key: "{{ kafka_controller_usm_agent_username }}"
      confluent.telemetry.exporter._usm.events.api.secret: "{{ kafka_controller_usm_agent_password }}"
  usm_agent_telemetry_auth_dummy:
    enabled: "{{ 'usm_agent' in groups and not usm_agent_basic_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.api.key: dummy
      confluent.telemetry.exporter._usm.api.secret: dummy
      confluent.telemetry.exporter._usm.events.api.key: dummy
      confluent.telemetry.exporter._usm.events.api.secret: dummy
  usm_agent_telemetry_ssl:
    enabled: "{{ 'usm_agent' in groups and usm_agent_ssl_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.https.ssl.protocol: TLSv1.2
      confluent.telemetry.exporter._usm.https.ssl.truststore.location: "{{ kafka_controller_truststore_path }}"
      confluent.telemetry.exporter._usm.https.ssl.truststore.password: "{{ kafka_controller_truststore_storepass }}"
      confluent.telemetry.exporter._usm.events.https.ssl.protocol: TLSv1.2
      confluent.telemetry.exporter._usm.events.https.ssl.truststore.location: "{{ kafka_controller_truststore_path }}"
      confluent.telemetry.exporter._usm.events.https.ssl.truststore.password: "{{ kafka_controller_truststore_storepass }}"
  usm_agent_telemetry_mtls:
    enabled: "{{ 'usm_agent' in groups and usm_agent_ssl_mutual_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.https.ssl.keystore.location: "{{ kafka_controller_keystore_path }}"
      confluent.telemetry.exporter._usm.https.ssl.keystore.password: "{{ kafka_controller_keystore_storepass }}"
      confluent.telemetry.exporter._usm.https.ssl.key.password: "{{ kafka_controller_keystore_keypass }}"
      confluent.telemetry.exporter._usm.events.https.ssl.keystore.location: "{{ kafka_controller_keystore_path }}"
      confluent.telemetry.exporter._usm.events.https.ssl.keystore.password: "{{ kafka_controller_keystore_storepass }}"
      confluent.telemetry.exporter._usm.events.https.ssl.key.password: "{{ kafka_controller_keystore_keypass }}"
  usm_client_metrics:
    enabled: "{{ 'usm_agent' in groups and usm_client_metrics_enabled }}"
    properties:
      confluent.telemetry.external.client.metrics.usm.push.enabled: "true"
      confluent.telemetry.external.client.metrics.push.enabled: "true"
      confluent.telemetry.external.client.metrics.delta.temporality: "true"
      confluent.telemetry.external.client.metrics.subscription.interval.ms.list: "60000"
      confluent.telemetry.external.client.metrics.subscription.metrics.list: "org.apache.kafka.consumer.fetch.manager.fetch.latency.avg,org.apache.kafka.consumer.connection.creation.total,org.apache.kafka.consumer.fetch.manager.fetch.total,org.apache.kafka.consumer.fetch.manager.bytes.consumed.rate,org.apache.kafka.consumer.request.total,org.apache.kafka.producer.bufferpool.wait.ratio,org.apache.kafka.producer.record.queue.time.avg,org.apache.kafka.producer.request.latency.avg,org.apache.kafka.producer.produce.throttle.time.avg,org.apache.kafka.producer.connection.creation.total,org.apache.kafka.producer.request.total,org.apache.kafka.producer.topic.byte.rate,org.apache.kafka.stream.client.state,org.apache.kafka.stream.thread.thread.state,org.apache.kafka.stream.state.size.all.mem.tables,org.apache.kafka.stream.state.estimate.num.keys,org.apache.kafka.stream.state.block.cache.usage,org.apache.kafka.stream.processor.node.record.e2e.latency.min,org.apache.kafka.stream.processor.node.record.e2e.latency.max,org.apache.kafka.stream.processor.node.record.e2e.latency.avg,org.apache.kafka.stream.thread.poll.ratio,org.apache.kafka.stream.thread.commit.ratio,org.apache.kafka.stream.thread.process.ratio,org.apache.kafka.stream.thread.punctuate.ratio,org.apache.kafka.stream.thread.process.records.avg"

kafka_controller_combined_properties: "{{kafka_controller_properties | confluent.platform.combine_properties}}"

kafka_controller_final_properties: "{{ kafka_controller_combined_properties | combine(kafka_controller_custom_properties) }}"

# A set of client properties against the controller listener for kafka health checks
kafka_controller_default_client_properties: "{{ kafka_controller_listeners['controller'] | confluent.platform.client_properties(kafka_controller_ssl_enabled, False, kafka_controller_ssl_mutual_auth_enabled, kafka_controller_sasl_protocol,
                            '', kafka_controller_pkcs12_truststore_path, kafka_controller_truststore_storepass, False, kafka_controller_pkcs12_keystore_path, kafka_controller_keystore_storepass, kafka_controller_keystore_keypass,
                            false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                            kerberos_kafka_controller_primary, kafka_controller_keytab_path, kafka_controller_kerberos_principal|default('kafka'),
                            false, kafka_controller_ldap_user, kafka_controller_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_controller_oauth_user, kafka_controller_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, true, kafka_controller_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_controller_third_party_oauth_client_assertion_config.controller)) }}"

kafka_controller_client_properties: "{{ kafka_controller_default_client_properties | combine(kafka_controller_custom_client_properties) }}"

#### Kafka Broker Variables ####
kafka_broker_service_name: "{{ 'confluent-server' if confluent_server_enabled|bool else 'confluent-kafka'}}"
kafka_broker_main_package: "{{ 'confluent-server' if confluent_server_enabled|bool else 'confluent-kafka'}}"
kafka_broker_default_user: cp-kafka
kafka_broker_default_group: confluent
kafka_broker_default_log_dir: /var/log/kafka
kafka_broker:
  server_start_file: "{{ binary_base_path }}/bin/kafka-server-start"
  config_file: "{{ (config_base_path, kafka_broker_config_prefix_path, 'server.properties') | path_join }}"
  client_config_file: "{{ (config_base_path, kafka_broker_config_prefix_path, 'client.properties') | path_join }}"
  systemd_file: "{{systemd_base_dir}}/{{kafka_broker_service_name}}.service"
  systemd_override: /etc/systemd/system/{{kafka_broker_service_name}}.service.d/override.conf
  log4j_file: "{{ (base_path, 'etc/kafka/log4j2.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, kafka_broker_config_prefix_path, 'kafka_server_jaas.conf') | path_join }}"
  rest_proxy_password_file: "{{ (config_base_path, kafka_broker_config_prefix_path, 'password.properties') | path_join }}"

mds_http_protocol: "{{ 'https' if kafka_broker_rest_ssl_enabled|bool else 'http' }}"
mds_tls_enabled: "{{true if 'https' in mds_bootstrap_server_urls else false}}"

kafka_broker_properties:
  defaults:
    enabled: true
    properties:
      group.initial.rebalance.delay.ms: 3000
      log.retention.check.interval.ms: 300000
      log.retention.hours: 168
      log.segment.bytes: 1073741824
      num.io.threads: 16
      num.network.threads: 8
      num.partitions: 1
      num.recovery.threads.per.data.dir: 2
      offsets.topic.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      socket.receive.buffer.bytes: 102400
      socket.request.max.bytes: 104857600
      socket.send.buffer.bytes: 102400
      transaction.state.log.min.isr: "{{ [ 2, kafka_broker_default_internal_replication_factor|int ] | min }}"
      transaction.state.log.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      advertised.listeners: "{% for listener in kafka_broker_listeners|dict2items %}{% if loop.index > 1%},{% endif %}{{ listener['value']['name'] }}://{{ listener['value']['hostname'] | default(hostvars[inventory_hostname]|confluent.platform.resolve_hostname) | confluent.platform.format_hostname  }}:{{ listener['value']['port'] }}{% endfor %}"
      confluent.ansible.managed: 'true'
      confluent.license.topic: _confluent-command
      confluent.license.topic.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      confluent.metadata.topic.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      confluent.balancer.topic.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      confluent.security.event.logger.exporter.kafka.topic.replicas: "{{audit_logs_destination_bootstrap_servers.split(',')|length if audit_logs_destination_enabled and rbac_enabled else kafka_broker_default_internal_replication_factor}}"
      confluent.support.metrics.enable: "true"
      confluent.support.customer.id: anonymous
      log.dirs: "/var/lib/kafka/data"
      kafka.rest.enable: "{{kafka_broker_rest_proxy_enabled|string|lower}}"
      inter.broker.listener.name: "{{kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['name']}}"
      confluent.cluster.link.metadata.topic.replication.factor: "{{kafka_broker_default_internal_replication_factor}}"
      process.roles: broker
  broker_on_controller:
    enabled: "{{kraft_enabled|bool}}"
    properties:
      controller.quorum.voters: "{{ kafka_controller_quorum_voters }}"
      controller.listener.names: "{{kafka_controller_listeners['controller']['name']}}"
      listener.security.protocol.map: "{% for listener in kafka_controller_listeners|dict2items %}{% if loop.index > 1%},{% endif %}{{ listener['value']['name'] }}:{{ listener['value'] | confluent.platform.kafka_protocol_defaults(kafka_controller_ssl_enabled, kafka_controller_sasl_protocol)}}{% endfor %},{% for listener in kafka_broker_listeners|dict2items %}{% if loop.index > 1%},{% endif %}{{ listener['value']['name'] }}:{{ listener['value'] | confluent.platform.kafka_protocol_defaults(ssl_enabled, sasl_protocol)}}{% endfor %}"
      listeners: "{% for listener in kafka_broker_listeners|dict2items %}{% if loop.index > 1%},{% endif %}{{ listener['value']['name'] }}://{{ listener['value']['ip'] | default('') }}:{{ listener['value']['port'] }}{% endfor %}"
  controller_sasl:
    enabled: "{{ kraft_enabled|bool and 'none' not in (kafka_controller_listeners['controller']['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol) }}"
    properties:
      sasl.mechanism.controller.protocol: "{{ (kafka_controller_sasl_protocol | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] }}"
  controller_listener:
    enabled: "{{ kraft_enabled|bool }}"
    properties: "{{ kafka_controller_listeners | confluent.platform.listener_properties(kafka_controller_ssl_enabled, fips_enabled, kafka_controller_ssl_client_authentication, principal_mapping_rules, kafka_controller_sasl_protocol,
                    kafka_broker_truststore_path, kafka_broker_truststore_storepass, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                    plain_jaas_config, kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'), kerberos_kafka_broker_primary,
                    sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password, rbac_enabled_public_pem_path, oauth_enabled, oauth_jwks_uri, oauth_expected_audience, oauth_sub_claim, rbac_enabled, false, false) }}"
  broker_id:
    enabled: "{{ inventory_hostname in groups.kafka_broker and not enable_node_id|bool }}"
    properties:
      broker.id: "{{ broker_id | default( groups.kafka_broker.index(inventory_hostname) + 1 ) if inventory_hostname in groups.kafka_broker else 0 }}"
  node_id:
    enabled: "{{ inventory_hostname in groups.kafka_broker and enable_node_id|bool }}"
    properties:
      node.id: "{{ (node_id if node_id is defined else (broker_id if broker_id is defined else groups.kafka_broker.index(inventory_hostname) + 1)) if inventory_hostname in groups.kafka_broker else 0 }}"

  ssl:
    enabled: "{{ kafka_controller_ssl_enabled|bool or
    kafka_broker_listeners | confluent.platform.ssl_required(ssl_enabled) or
    kafka_broker_rest_ssl_enabled|bool or
    mds_broker_listener.ssl_enabled|bool or
    mds_tls_enabled|bool or
    ( kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_ssl_enabled ) }}"
    properties:
      ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
      ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      ssl.key.password: "{{kafka_broker_keystore_storepass}}"
  principal_mapping_rules:
    enabled: "{{ssl_enabled}}"
    properties:
      # This config is needed to get correct super user in set_principal.yml
      # Only setting rules on listener or mds or erp isnt enough
      ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  fips:
    enabled: "{{fips_enabled}}"
    properties:
      enable.fips: 'true'
      enable.fips.mode: "{{fips_mode}}"
      security.providers: io.confluent.kafka.security.fips.provider.BcFipsProviderCreator,io.confluent.kafka.security.fips.provider.BcFipsJsseProviderCreator
      ssl.keymanager.algorithm: PKIX
      ssl.trustmanager.algorithm: PKIX
      ssl.keystore.type: BCFKS
      ssl.truststore.type: BCFKS
      ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
      ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      ssl.key.password: "{{kafka_broker_keystore_storepass}}"
      ssl.enabled.protocols: TLSv1.2,TLSv1.3
  inter_broker_sasl:
    enabled: "{{ 'none' not in (kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol) }}"
    properties:
      sasl.mechanism.inter.broker.protocol: "{{ (kafka_broker_listeners[kafka_broker_inter_broker_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] }}"
  sasl_enabled:
    enabled: "{{ kafka_sasl_enabled_mechanisms|length > 0 }}"
    properties:
      sasl.enabled.mechanisms: "{% for mechanism in kafka_sasl_enabled_mechanisms %}{% if loop.index > 1%},{% endif %}{{ mechanism|upper }}{% endfor %}"
  sasl_gssapi:
    enabled: "{{ 'GSSAPI' in kafka_sasl_enabled_mechanisms }}"
    properties:
      sasl.kerberos.service.name: "{{kerberos_kafka_broker_primary}}"
  sr:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups }}"
    properties:
      confluent.schema.registry.url: "{{schema_registry_url}}"
  sr_mtls:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_ssl_client_authentication != 'none' }}"
    properties:
      confluent.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
      confluent.ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      confluent.ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      confluent.ssl.key.password: "{{kafka_broker_keystore_keypass}}"
  sr_ssl:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_ssl_enabled }}"
    properties:
      confluent.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
      confluent.ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      confluent.ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      confluent.ssl.key.password: "{{kafka_broker_keystore_keypass}}"
  sr_ssl_fips:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_ssl_enabled and fips_enabled }}"
    properties:
      confluent.ssl.keymanager.algorithm: PKIX
      confluent.ssl.trustmanager.algorithm: PKIX
      confluent.ssl.keystore.type: BCFKS
      confluent.ssl.truststore.type: BCFKS
  sr_rbac_ldap:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and rbac_enabled and schema_registry_auth_mode == 'ldap' }}"
    properties:
      confluent.basic.auth.credentials.source: USER_INFO
      confluent.basic.auth.user.info: "{{schema_registry_ldap_user | default('sr') }}:{{schema_registry_ldap_password | default('pass')}}"
  sr_oauth:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled }}"
    properties:
      confluent.bearer.auth.issuer.endpoint.url: "{{ oauth_token_uri }}"
      confluent.bearer.auth.client.id: "{{ schema_registry_oauth_user }}"
  sr_oauth_secret:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and (not schema_registry_oauth_client_assertion_enabled) }}"
    properties:
      confluent.bearer.auth.credentials.source: OAUTHBEARER
      confluent.bearer.auth.client.secret: "{{ schema_registry_oauth_password }}"
  sr_oauth_scope:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and oauth_groups_scope!='none' and (not schema_registry_oauth_client_assertion_enabled) }}"
    properties:
      confluent.bearer.auth.scope: "{{ oauth_groups_scope }}"
  sr_oauth_ssl:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and idp_self_signed}}"
    properties:
      confluent.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  sr_oauth_third_party_client_assertion:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path != 'none' }}"
    properties:
      confluent.bearer.assertion.file: "{{ kafka_broker_third_party_oauth_client_assertion_config.schema_registry }}"
  sr_oauth_local_client_assertion:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' }}"
    properties:
      confluent.bearer.auth.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.bearer.assertion.claim.iss: "{{schema_registry_oauth_client_assertion_issuer}}"
      confluent.bearer.assertion.claim.sub: "{{schema_registry_oauth_client_assertion_sub}}"
      confluent.bearer.assertion.claim.aud: "{{schema_registry_oauth_client_assertion_audience}}"
      confluent.bearer.assertion.private.key.file: "{{ kafka_broker_sr_oauth_client_assertion_private_key_file_dest_path }}"
  sr_oauth_local_client_assertion_template_file:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not ldap_with_oauth_enabled) and schema_registry_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.bearer.assertion.template.file: "{{ kafka_broker_sr_oauth_client_assertion_template_file_dest_path }}"
  sr_oauth_local_client_assertion_passphrase:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not ldap_with_oauth_enabled) and schema_registry_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.bearer.assertion.private.key.passphrase: "{{schema_registry_oauth_client_assertion_private_key_passphrase}}"
  sr_oauth_local_client_assertion_jti_include:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not ldap_with_oauth_enabled) and schema_registry_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.bearer.assertion.claim.jti.include: "{{schema_registry_oauth_client_assertion_jti_include|bool}}"
  sr_oauth_local_client_assertion_nbf_include:
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not ldap_with_oauth_enabled) and schema_registry_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.bearer.assertion.claim.nbf.include: "{{schema_registry_oauth_client_assertion_nbf_include}}"
  sr_basic:
    # Should not turn on basic auth if rbac is enabled
    enabled: "{{ kafka_broker_schema_validation_enabled and 'schema_registry' in groups and schema_registry_authentication_type == 'basic'}}"
    properties:
      confluent.basic.auth.credentials.source: USER_INFO
      confluent.basic.auth.user.info: "{{schema_registry_basic_users_final.admin.principal}}:{{schema_registry_basic_users_final.admin.password}}"
  rbac:
    enabled: "{{ rbac_enabled }}"
    properties:
      authorizer.class.name: io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
      confluent.authorizer.access.rule.providers: CONFLUENT,KRAFT_ACL
      super.users: >-
        {{
           ('ldap' in auth_mode) | ternary("User:" + mds_super_user|default('mds'), "") +
           (auth_mode == 'ldap_with_oauth') | ternary(";", "") +
           ('oauth' in auth_mode) | ternary("User:" + oauth_superuser_principal, "")
        }}
  rbac_mds:
    enabled: "{{ rbac_enabled and not external_mds_enabled }}"
    properties:
      confluent.metadata.server.advertised.listeners: "{{mds_http_protocol}}://{{ mds_advertised_listener_hostname | default(hostvars[inventory_hostname]|confluent.platform.resolve_hostname, True) |  confluent.platform.format_hostname}}:{{mds_port}}"
      confluent.metadata.server.listeners: "{{mds_http_protocol}}://0.0.0.0:{{mds_port}}"
      confluent.metadata.server.token.max.lifetime.ms: 3600000
      confluent.metadata.server.token.key.path: "{{rbac_enabled_private_pem_path}}"
      confluent.metadata.server.token.signature.algorithm: RS256
      confluent.metadata.server.authentication.method: BEARER
  rbac_mds_oauth:
    enabled: "{{ rbac_enabled and not external_mds_enabled and oauth_enabled }}"
    properties:
      confluent.metadata.server.user.store: "{{ 'LDAP_WITH_OAUTH' if ldap_with_oauth_enabled|bool else 'OAUTH' }}"
      confluent.metadata.server.oauthbearer.jwks.endpoint.url: "{{oauth_jwks_uri}}"
      confluent.metadata.server.oauthbearer.expected.issuer: "{{oauth_issuer_url}}"
      confluent.metadata.server.oauthbearer.sub.claim.name: "{{oauth_sub_claim}}"
  rbac_mds_oauth_audience:
    enabled: "{{ rbac_enabled and not external_mds_enabled and oauth_enabled and oauth_expected_audience!= 'none' }}"
    properties:
      confluent.metadata.server.oauthbearer.expected.audience: "{{oauth_expected_audience}}"
  rbac_mds_oauth_claim:
    enabled: "{{ rbac_enabled and not external_mds_enabled and oauth_enabled and oauth_groups_claim != 'none'}}"
    properties:
      confluent.metadata.server.oauthbearer.groups.claim.name: "{{oauth_groups_claim}}"
  rbac_mds_ssl:
    enabled: "{{ rbac_enabled and not external_mds_enabled and oauth_enabled and mds_ssl_enabled and mds_ssl_client_authentication == 'none' }}"
    properties:
      confluent.metadata.server.ssl.endpoint.identification.algorithm: https
  rbac_mds_ldap:
    enabled: "{{ rbac_enabled and not external_mds_enabled and 'ldap' in auth_mode }}"
    # For backwards compatibility, need to make sure ldap_config var is honored
    properties: "{{ ldap_config | default('') | confluent.platform.split_newline_to_dict }}"
  rbac_mds_rest_ssl:
    enabled: "{{ rbac_enabled and not external_mds_enabled and kafka_broker_rest_ssl_enabled }}"
    properties:
      confluent.metadata.server.ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      confluent.metadata.server.ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      confluent.metadata.server.ssl.key.password: "{{kafka_broker_keystore_keypass}}"
      confluent.metadata.server.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.metadata.server.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  rbac_mds_principal_mapping:
    enabled: "{{ rbac_enabled and not external_mds_enabled and principal_mapping_rules != '' and mds_ssl_enabled|bool }}"
    properties:
      confluent.metadata.server.auth.ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  rbac_mds_mtls_only:
    enabled: "{{ rbac_enabled and not external_mds_enabled and mds_ssl_enabled|bool and auth_mode == 'mtls' and not mds_file_based_user_store_enabled|bool }}"
    properties:
      confluent.metadata.server.user.store: NONE
  rbac_mds_file_store:
    enabled: "{{ rbac_enabled and not external_mds_enabled and mds_file_based_user_store_enabled|bool }}"
    properties:
      confluent.metadata.server.user.store: FILE
      confluent.metadata.server.user.store.file.path: "{{ mds_file_based_user_store_dest_path }}"
  rbac_mds_client_authentication:
    enabled: "{{ rbac_enabled and not external_mds_enabled and mds_ssl_enabled|bool and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.server.ssl.client.authentication: "{{ mds_ssl_client_authentication | upper }}"
  rbac_mds_impersonation_super_users:
    enabled: "{{ rbac_enabled|bool and not external_mds_enabled|bool and impersonation_super_users|length > 0 }}"
    properties:
      confluent.metadata.server.impersonation.super.users: "{{impersonation_super_users | map('regex_replace', '^', 'User:') | join(';')}}"
  rbac_mds_impersonation_protected_users:
    enabled: "{{ rbac_enabled|bool and not external_mds_enabled|bool and impersonation_protected_users|length > 0 }}"
    properties:
      confluent.metadata.server.impersonation.protected.users: "{{impersonation_protected_users | map('regex_replace','^', 'User:') | join(';')}}"
  rbac_mds_ssl_fips:
    enabled: "{{ rbac_enabled and not external_mds_enabled and kafka_broker_rest_ssl_enabled and fips_enabled }}"
    properties:
      confluent.metadata.server.ssl.keymanager.algorithm: PKIX
      confluent.metadata.server.ssl.trustmanager.algorithm: PKIX
      confluent.metadata.server.ssl.keystore.type: BCFKS
      confluent.metadata.server.ssl.truststore.type: BCFKS
      confluent.metadata.server.http2.enabled: false
  rbac_mds_sso:
    enabled: "{{ rbac_enabled and not external_mds_enabled and sso_mode != 'none' }}"
    properties:
      confluent.metadata.server.sso.mode: "{{ sso_mode }}"
      confluent.oidc.idp.groups.claim.name: "{{ sso_groups_claim }}"
      confluent.oidc.idp.sub.claim.name: "{{ sso_sub_claim }}"
      confluent.oidc.idp.issuer: "{{ sso_issuer_url }}"
      confluent.oidc.idp.jwks.endpoint.uri: "{{ sso_jwks_uri }}"
      confluent.oidc.idp.authorize.base.endpoint.uri: "{{ sso_authorize_uri }}"
      confluent.oidc.idp.token.base.endpoint.uri: "{{ sso_token_uri }}"
      confluent.oidc.idp.client.id: "{{ sso_client_id }}"
      confluent.oidc.idp.client.secret: "{{ sso_client_password }}"
      confluent.oidc.idp.refresh.token.enabled: "{{ sso_refresh_token }}"
      confluent.oidc.session.token.expiry.ms: 900000
      confluent.oidc.session.max.timeout.ms: 21600000
  rbac_mds_sso_scope:
    enabled: "{{ rbac_enabled and not external_mds_enabled and sso_mode != 'none' and sso_groups_scope != 'none' }}"
    properties:
      confluent.oidc.idp.groups.claim.scope: "{{ sso_groups_scope }}"
  rbac_mds_sso_ssl:
    enabled: "{{ rbac_enabled and (not external_mds_enabled) and sso_mode != 'none' and sso_idp_cert_path != ''}}"
    properties:
      confluent.metadata.server.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.metadata.server.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  rbac_external_mds:
    enabled: "{{rbac_enabled and external_mds_enabled}}"
    properties:
      confluent.metadata.bootstrap.servers: "{{mds_broker_bootstrap_servers}}"
  rbac_external_mds_client:
    enabled: "{{rbac_enabled and external_mds_enabled}}"
    properties: "{{ mds_broker_listener | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                            'confluent.metadata.', kafka_broker_truststore_path, kafka_broker_truststore_storepass, False, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                            false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                            kerberos_kafka_broker_primary, kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'),
                            false, kafka_broker_ldap_user, kafka_broker_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_broker_oauth_user, kafka_broker_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_broker_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_broker_third_party_oauth_client_assertion_config.mds)) }}"
  rbac_external_mds_oauth_client:
    enabled: "{{ oauth_enabled and rbac_enabled and external_mds_enabled and idp_self_signed }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  rbac_cli_sso:
    enabled: "{{sso_cli}}"
    properties:
      confluent.oidc.idp.device.authorization.endpoint.uri: "{{ sso_device_authorization_uri }}"
  rbac_oauth_ssl:
    enabled: "{{ oauth_enabled and rbac_enabled and (not external_mds_enabled) and idp_self_signed }}"
    properties:
      confluent.metadata.server.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.metadata.server.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  embedded_rest_proxy:
    # Do not need duplicating confluent.metadata.server and confluent.http.server config, rely on mds configs when kafka is the mds
    enabled: "{{ kafka_broker_rest_proxy_enabled and (not rbac_enabled or (rbac_enabled and external_mds_enabled)) }}"
    properties:
      confluent.http.server.advertised.listeners: "{{mds_http_protocol}}://{{ mds_advertised_listener_hostname | default(hostvars[inventory_hostname] | confluent.platform.resolve_hostname , True)| confluent.platform.format_hostname }}:{{mds_port}}"
      confluent.http.server.listeners: "{{mds_http_protocol}}://0.0.0.0:{{mds_port}}"
  embedded_rest_proxy_ssl:
    enabled: "{{ kafka_broker_rest_proxy_enabled and (not rbac_enabled or (rbac_enabled and external_mds_enabled)) and kafka_broker_rest_ssl_enabled }}"
    properties:
      confluent.http.server.ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      confluent.http.server.ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      confluent.http.server.ssl.key.password: "{{kafka_broker_keystore_keypass}}"
      confluent.http.server.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.http.server.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  embedded_rest_proxy_ssl_fips:
    enabled: "{{ kafka_broker_rest_proxy_enabled and kafka_broker_rest_ssl_enabled and fips_enabled}}"
    properties:
      confluent.http.server.ssl.keymanager.algorithm: PKIX
      confluent.http.server.ssl.trustmanager.algorithm: PKIX
      confluent.http.server.ssl.keystore.type: BCFKS
      confluent.http.server.ssl.truststore.type: BCFKS
      confluent.http.server.http2.enabled: false
  embedded_rest_proxy_fips:
    enabled: "{{ kafka_broker_rest_proxy_enabled and fips_enabled }}"
    properties:
      confluent.metadata.server.sni.host.check.enabled: false #this is ERP property and not MDS property
  embedded_rest_proxy_mtls:
    enabled: "{{ kafka_broker_rest_proxy_enabled and kafka_broker_rest_ssl_enabled and erp_ssl_client_authentication != 'none' }}"
    properties:
      kafka.rest.ssl.client.authentication: "{{ erp_ssl_client_authentication | upper }}"
  embedded_rest_proxy_basic:
    enabled: "{{ kafka_broker_rest_proxy_enabled and (not rbac_enabled or (rbac_enabled and external_mds_enabled)) and kafka_broker_rest_proxy_authentication_type == 'basic' }}"
    properties:
      kafka.rest.resource.extension.class: io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
      kafka.rest.confluent.rest.auth.propogate.method: JETTY_AUTH
      kafka.rest.authentication.method: BASIC
      kafka.rest.authentication.realm: KafkaRest
      kafka.rest.authentication.roles: "{{ kafka_broker_rest_proxy_basic_users | confluent.platform.get_roles | unique | join(',') }}"
  embedded_rest_proxy_client_bootstrap:
    enabled: "{{ kafka_broker_rest_proxy_enabled }}"
    properties:
      kafka.rest.bootstrap.servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_broker_rest_proxy_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_broker_rest_proxy_listener_name]['port']}}"
  embedded_rest_proxy_client:
    enabled: "{{ kafka_broker_rest_proxy_enabled }}"
    properties: "{{ kafka_broker_listeners[kafka_broker_rest_proxy_listener_name] | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                    'kafka.rest.client.', kafka_broker_truststore_path, kafka_broker_truststore_storepass, False, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_broker_primary, kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'),
                    true, kafka_broker_ldap_user, kafka_broker_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_broker_oauth_user, kafka_broker_oauth_password, oauth_groups_scope, oauth_token_uri, false, false, kafka_broker_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_broker_third_party_oauth_client_assertion_config.embedded_rest_proxy)) }}"
  embedded_rest_proxy_rbac:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled }}"
    properties:
      kafka.rest.kafka.rest.resource.extension.class: io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
      kafka.rest.rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      kafka.rest.public.key.path: "{{rbac_enabled_public_pem_path}}"
      kafka.rest.confluent.metadata.bootstrap.server.urls: "{{mds_bootstrap_server_urls}}"
  embedded_rest_proxy_rbac_ldap:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and auth_mode == 'ldap' and (not kafka_broker_rest_proxy_mds_cert_auth_only) }}"
    properties:
      kafka.rest.confluent.metadata.basic.auth.user.info: "{{kafka_broker_ldap_user | default('kafka') }}:{{kafka_broker_ldap_password | default('pass')}}"
      kafka.rest.confluent.metadata.http.auth.credentials.provider: BASIC
  embedded_rest_proxy_rbac_oauth:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and (not kafka_broker_rest_proxy_mds_cert_auth_only) }}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.login.client.id: "{{kafka_broker_oauth_user}}"
      kafka.rest.confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
  embedded_rest_proxy_rbac_oauth_secret:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and (not kafka_broker_rest_proxy_mds_cert_auth_only) and (not kafka_broker_oauth_client_assertion_enabled) }}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.login.client.secret: "{{kafka_broker_oauth_password}}"
  embedded_rest_proxy_rbac_oauth_scope:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and oauth_groups_scope!='none' and (not kafka_broker_rest_proxy_mds_cert_auth_only) and (not kafka_broker_oauth_client_assertion_enabled) }}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.login.oauth.scope: "{{ oauth_groups_scope }}"
  embedded_rest_proxy_rbac_oauth_third_party_client_assertion:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and kafka_broker_oauth_client_assertion_enabled and kafka_broker_oauth_client_assertion_file_base_path != 'none' and (not kafka_broker_rest_proxy_mds_cert_auth_only)}}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.assertion.file: "{{ kafka_broker_third_party_oauth_client_assertion_config.embedded_rest_proxy_rbac_oauth }}"
  embedded_rest_proxy_rbac_oauth_local_client_assertion:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and kafka_broker_oauth_client_assertion_enabled and kafka_broker_oauth_client_assertion_file_base_path == 'none' and (not kafka_broker_rest_proxy_mds_cert_auth_only)}}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.assertion.claim.iss: "{{kafka_broker_oauth_client_assertion_issuer}}"
      kafka.rest.confluent.metadata.oauthbearer.assertion.claim.sub: "{{kafka_broker_oauth_client_assertion_sub}}"
      kafka.rest.confluent.metadata.oauthbearer.assertion.claim.aud: "{{kafka_broker_oauth_client_assertion_audience}}"
      kafka.rest.confluent.metadata.oauthbearerassertion.private.key.file: "{{ kafka_broker_oauth_client_assertion_private_key_file_dest_path }}"
  embedded_rest_proxy_rbac_oauth_local_client_assertion_template:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and kafka_broker_oauth_client_assertion_enabled and kafka_broker_oauth_client_assertion_file_base_path == 'none' and kafka_broker_oauth_client_assertion_template_file != 'none' and (not kafka_broker_rest_proxy_mds_cert_auth_only)}}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.assertion.template.file: "{{ kafka_broker_oauth_client_assertion_template_file_dest_path }}"
  embedded_rest_proxy_rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and kafka_broker_oauth_client_assertion_enabled and kafka_broker_oauth_client_assertion_file_base_path == 'none' and kafka_broker_oauth_client_assertion_private_key_passphrase != 'none' and (not kafka_broker_rest_proxy_mds_cert_auth_only)}}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{kafka_broker_oauth_client_assertion_private_key_passphrase}}"
  embedded_rest_proxy_rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and kafka_broker_oauth_client_assertion_enabled and kafka_broker_oauth_client_assertion_file_base_path == 'none' and kafka_broker_oauth_client_assertion_jti_include != 'none' and (not kafka_broker_rest_proxy_mds_cert_auth_only)}}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{kafka_broker_oauth_client_assertion_jti_include}}"
  embedded_rest_proxy_rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{ kafka_broker_rest_proxy_enabled and rbac_enabled and oauth_enabled and kafka_broker_oauth_client_assertion_enabled and kafka_broker_oauth_client_assertion_file_base_path == 'none' and kafka_broker_oauth_client_assertion_nbf_include != 'none' and (not kafka_broker_rest_proxy_mds_cert_auth_only)}}"
    properties:
      kafka.rest.confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{kafka_broker_oauth_client_assertion_nbf_include}}"
  embedded_rest_proxy_oauth:
    enabled: "{{ kafka_broker_rest_proxy_enabled and oauth_enabled }}"
    properties:
      kafka.rest.rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      kafka.rest.oauthbearer.jwks.endpoint.url: "{{ oauth_jwks_uri }}"
      kafka.rest.oauthbearer.sub.claim.name: "{{ oauth_sub_claim }}"
      kafka.rest.oauthbearer.expected.issuer: "{{ oauth_issuer_url }}"
  embedded_rest_proxy_oauth_audience:
    enabled: "{{ kafka_broker_rest_proxy_enabled and oauth_enabled and oauth_expected_audience!='none' }}"
    properties:
      kafka.rest.oauthbearer.expected.audience: "{{ oauth_expected_audience }}"
  embedded_rest_proxy_oauth_claim:
    enabled: "{{ kafka_broker_rest_proxy_enabled and oauth_enabled and oauth_groups_claim!='none' }}"
    properties:
      kafka.rest.oauthbearer.groups.claim.name: "{{ oauth_groups_claim }}"
  embedded_rest_proxy_oauth_ssl:
    enabled: "{{ kafka_broker_rest_proxy_enabled and oauth_enabled and (not rbac_enabled or (rbac_enabled and external_mds_enabled)) and idp_self_signed}}"
    properties:
      confluent.http.server.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.http.server.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  embedded_rest_proxy_rbac_ssl:
    enabled: "{{rbac_enabled and mds_tls_enabled}}"
    properties:
      kafka.rest.confluent.metadata.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      kafka.rest.confluent.metadata.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
  embedded_rest_proxy_rbac_mtls:
    enabled: "{{rbac_enabled and mds_tls_enabled and mds_ssl_client_authentication != 'none'}}"
    # truststore is getting set in above ssl section
    properties:
      kafka.rest.confluent.metadata.ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      kafka.rest.confluent.metadata.ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      kafka.rest.confluent.metadata.ssl.key.password: "{{kafka_broker_keystore_keypass}}"
  embedded_rest_proxy_principal_mapping_rules:
    enabled: "{{ kafka_broker_rest_proxy_enabled and principal_mapping_rules != '' }}"
    properties:
      kafka.rest.auth.ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  oauth_claim:
    enabled: "{{ oauth_enabled and oauth_groups_claim != 'none'}}"
    properties:
      confluent.oauth.groups.claim.name: "{{oauth_groups_claim}}"
  listeners:
    enabled: true
    properties: "{{ kafka_broker_listeners | confluent.platform.listener_properties(ssl_enabled, fips_enabled, ssl_client_authentication, principal_mapping_rules, sasl_protocol,
                    kafka_broker_truststore_path, kafka_broker_truststore_storepass, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                    plain_jaas_config, kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'), kerberos_kafka_broker_primary,
                    sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password, rbac_enabled_public_pem_path, oauth_enabled, oauth_jwks_uri, oauth_expected_audience, oauth_sub_claim, rbac_enabled, false, idp_self_signed) }}"
  controlplane_listener:
    enabled: "{{ kafka_broker_configure_control_plane_listener|bool and kafka_broker_configure_multiple_listeners|bool and not kraft_enabled|bool}}"
    properties:
      control.plane.listener.name: "{{(kafka_broker_listeners[kafka_broker_control_plane_listener_name]|default(kafka_broker_listeners[kafka_broker_inter_broker_listener_name]))['name']}}"
  metrics_reporter_for_control_center_next_gen:
    enabled: "{{ kafka_broker_metrics_reporter_for_control_center_next_gen_enabled|bool }}"
    properties:
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter,org.apache.kafka.common.metrics.JmxReporter
      confluent.telemetry.exporter._c3.type: "http"
      confluent.telemetry.exporter._c3.enabled: "true"
      confluent.telemetry.exporter._c3.metrics.include: "io.confluent.kafka.server.request.(?!.*delta).*|io.confluent.kafka.server.server.broker.state|io.confluent.kafka.server.replica.manager.leader.count|io.confluent.kafka.server.request.queue.size|io.confluent.kafka.server.broker.topic.failed.produce.requests.rate.1.min|io.confluent.kafka.server.tier.archiver.total.lag|io.confluent.kafka.server.request.total.time.ms.p99|io.confluent.kafka.server.broker.topic.failed.fetch.requests.rate.1.min|io.confluent.kafka.server.broker.topic.total.fetch.requests.rate.1.min|io.confluent.kafka.server.partition.caught.up.replicas.count|io.confluent.kafka.server.partition.observer.replicas.count|io.confluent.kafka.server.tier.tasks.num.partitions.in.error|io.confluent.kafka.server.broker.topic.bytes.out.rate.1.min|io.confluent.kafka.server.request.total.time.ms.p95|io.confluent.kafka.server.controller.active.controller.count|io.confluent.kafka.server.session.expire.listener.zookeeper.disconnects.total|io.confluent.kafka.server.request.total.time.ms.p999|io.confluent.kafka.server.controller.active.broker.count|io.confluent.kafka.server.request.handler.pool.request.handler.avg.idle.percent.rate.1.min|io.confluent.kafka.server.session.expire.listener.zookeeper.disconnects.rate.1.min|io.confluent.kafka.server.controller.unclean.leader.elections.rate.1.min|io.confluent.kafka.server.replica.manager.partition.count|io.confluent.kafka.server.controller.unclean.leader.elections.total|io.confluent.kafka.server.partition.replicas.count|io.confluent.kafka.server.broker.topic.total.produce.requests.rate.1.min|io.confluent.kafka.server.controller.offline.partitions.count|io.confluent.kafka.server.socket.server.network.processor.avg.idle.percent|io.confluent.kafka.server.partition.under.replicated|io.confluent.kafka.server.log.log.start.offset|io.confluent.kafka.server.log.tier.size|io.confluent.kafka.server.log.size|io.confluent.kafka.server.tier.fetcher.bytes.fetched.total|io.confluent.kafka.server.request.total.time.ms.p50|io.confluent.kafka.server.tenant.consumer.lag.offsets|io.confluent.kafka.server.session.expire.listener.zookeeper.expires.rate.1.min|io.confluent.kafka.server.log.log.end.offset|io.confluent.kafka.server.broker.topic.bytes.in.rate.1.min|io.confluent.kafka.server.partition.under.min.isr|io.confluent.kafka.server.partition.in.sync.replicas.count|io.confluent.telemetry.http.exporter.batches.dropped|io.confluent.telemetry.http.exporter.items.total|io.confluent.telemetry.http.exporter.items.succeeded|io.confluent.telemetry.http.exporter.send.time.total.millis|io.confluent.kafka.server.controller.leader.election.rate.(?!.*delta).*|io.confluent.telemetry.http.exporter.batches.failed|io.confluent.kafka.server.socket_server.connections|org.apache.kafka.consumer.(fetch.manager.fetch.latency.avg|connection.creation.total|fetch.manager.fetch.total|fetch.manager.bytes.consumed.rate)|org.apache.kafka.producer.(bufferpool.wait.ratio|request.total|topic.byte.rate)|io.confluent.kafka.server.controller.fenced.broker.count"
      confluent.telemetry.exporter._c3.client.base.url: "{{control_center_next_gen_dependency_prometheus_url}}/api/v1/otlp"
      confluent.telemetry.exporter._c3.client.compression: gzip
      confluent.telemetry.exporter._c3.api.key: "{{kafka_broker_telemetry_control_center_next_gen_user}}"
      confluent.telemetry.exporter._c3.api.secret: "{{kafka_broker_telemetry_control_center_next_gen_password}}"
      confluent.telemetry.exporter._c3.buffer.pending.batches.max: 80
      confluent.telemetry.exporter._c3.buffer.batch.items.max: 4000
      confluent.telemetry.exporter._c3.buffer.inflight.submissions.max: 10
      confluent.telemetry.metrics.collector.interval.ms: 60000
      confluent.telemetry.remoteconfig._confluent.enabled: false
      confluent.consumer.lag.emitter.enabled: "true"
  metrics_reporter_for_control_center_next_gen_client:
    enabled: "{{control_center_next_gen_dependency_prometheus_ssl_enabled}}"
    properties:
      confluent.telemetry.exporter._c3.https.ssl.protocol: TLSv1.2
      confluent.telemetry.exporter._c3.https.ssl.truststore.location: "{{kafka_broker_truststore_path}}"
      confluent.telemetry.exporter._c3.https.ssl.truststore.password: "{{kafka_broker_truststore_storepass}}"
      # If FIPS set to BCFKS else JKS
      confluent.telemetry.exporter._c3.https.ssl.truststore.type:  "{% if fips_enabled|bool %}BCFKS{% else %}JKS{% endif %}"
  metrics_reporter_for_control_center_next_gen_client_mtls:
    enabled: "{{control_center_next_gen_dependency_prometheus_mtls_enabled}}"
    properties:
      confluent.telemetry.exporter._c3.https.ssl.keystore.location: "{{kafka_broker_keystore_path}}"
      confluent.telemetry.exporter._c3.https.ssl.keystore.password: "{{kafka_broker_keystore_storepass}}"
      confluent.telemetry.exporter._c3.https.ssl.key.password: "{{kafka_broker_keystore_keypass}}"
      # If FIPS set to BCFKS else JKS
      confluent.telemetry.exporter._c3.https.ssl.keystore.type: "{% if fips_enabled|bool %}BCFKS{% else %}JKS{% endif %}"
  metrics_reporter_client:
    enabled: "{{ kafka_broker_metrics_reporter_enabled|bool }}"
    properties: "{{ kafka_broker_listeners[kafka_broker_inter_broker_listener_name] | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.metrics.reporter.', kafka_broker_truststore_path, kafka_broker_truststore_storepass, False, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_broker_primary, kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'),
                    false, kafka_broker_ldap_user, kafka_broker_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_broker_oauth_user, kafka_broker_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_broker_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_broker_third_party_oauth_client_assertion_config.metrics_reporter)) }}"
  telemetry:
    enabled: "{{kafka_broker_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
  telemetry_proxy:
    enabled: "{{kafka_broker_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{kafka_broker_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{kafka_broker_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"
  audit_logs_destination:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled}}"
    properties:
      confluent.security.event.logger.exporter.kafka.bootstrap.servers: "{{audit_logs_destination_bootstrap_servers}}"
      confluent.security.event.logger.exporter.kafka.topic.create: 'false'
  audit_logs_destination_client:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled}}"
    properties: "{{ audit_logs_destination_listener | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.security.event.logger.exporter.kafka.', kafka_broker_truststore_path, kafka_broker_truststore_storepass, False, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_broker_primary|default('kafka'), kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'),
                    false, 'user', 'pass', mds_bootstrap_server_urls, oauth_enabled, kafka_broker_oauth_user, kafka_broker_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_broker_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_broker_third_party_oauth_client_assertion_config.audit_logs_destination_client)) }}"
  audit_logs_destination_admin:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled and not external_mds_enabled}}"
    properties:
      confluent.security.event.logger.destination.admin.bootstrap.servers: "{{audit_logs_destination_bootstrap_servers}}"
  audit_logs_destination_admin_client:
    enabled: "{{audit_logs_destination_enabled and rbac_enabled and not external_mds_enabled}}"
    properties: "{{ audit_logs_destination_listener | confluent.platform.client_properties(ssl_enabled, fips_enabled, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.security.event.logger.destination.admin.', kafka_broker_truststore_path, kafka_broker_truststore_storepass, False, kafka_broker_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                    false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                    kerberos_kafka_broker_primary|default('kafka'), kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'),
                    false, 'user', 'pass', mds_bootstrap_server_urls, oauth_enabled, kafka_broker_oauth_user, kafka_broker_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_broker_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_broker_third_party_oauth_client_assertion_config.audit_logs_destination_admin_client)) }}"
  usm_agent_telemetry:
    enabled: "{{ 'usm_agent' in groups }}"
    properties:
      confluent.telemetry.exporter._usm.enabled: true
      confluent.telemetry.exporter._usm.client.base.url: "{{ usm_agent_url }}"
      confluent.telemetry.exporter._usm.type: http
      confluent.telemetry.exporter._usm.buffer.batch.items.max: 4000
      confluent.telemetry.exporter._usm.buffer.inflight.submissions.max: 10
      confluent.telemetry.exporter._usm.buffer.pending.batches.max: 80
      confluent.client.topic.metrics.manager: org.apache.kafka.server.metrics.PlatformClientTopicMetricsManager
      confluent.consumer.lag.emitter.enabled: true
      confluent.consumer.lag.calculator.enabled: "true"
      confluent.consumer.group.status.enabled: "true"
      confluent.consumer.lag.calculator.empty.lag.retention.ms: 86400000
  usm_agent_telemetry_auth_basic:
    enabled: "{{ 'usm_agent' in groups and usm_agent_basic_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.api.key: "{{ kafka_broker_usm_agent_username }}"
      confluent.telemetry.exporter._usm.api.secret: "{{ kafka_broker_usm_agent_password }}"
  usm_agent_telemetry_auth_dummy:
    enabled: "{{ 'usm_agent' in groups and not usm_agent_basic_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.api.key: dummy
      confluent.telemetry.exporter._usm.api.secret: dummy
  usm_agent_telemetry_ssl:
    enabled: "{{ 'usm_agent' in groups and usm_agent_ssl_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.https.ssl.protocol: TLSv1.2
      confluent.telemetry.exporter._usm.https.ssl.truststore.location: "{{ kafka_broker_truststore_path }}"
      confluent.telemetry.exporter._usm.https.ssl.truststore.password: "{{ kafka_broker_truststore_storepass }}"
  usm_agent_telemetry_mtls:
    enabled: "{{ 'usm_agent' in groups and usm_agent_ssl_mutual_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.https.ssl.keystore.location: "{{ kafka_broker_keystore_path }}"
      confluent.telemetry.exporter._usm.https.ssl.keystore.password: "{{ kafka_broker_keystore_storepass }}"
      confluent.telemetry.exporter._usm.https.ssl.key.password: "{{ kafka_broker_keystore_keypass }}"
  usm_client_metrics:
    enabled: "{{ 'usm_agent' in groups and usm_client_metrics_enabled }}"
    properties:
      confluent.telemetry.external.client.metrics.usm.push.enabled: "true"
      confluent.telemetry.external.client.metrics.push.enabled: "true"
      confluent.telemetry.external.client.metrics.delta.temporality: "true"
      confluent.telemetry.external.client.metrics.subscription.interval.ms.list: "60000"
      confluent.telemetry.external.client.metrics.subscription.metrics.list: "org.apache.kafka.consumer.fetch.manager.fetch.latency.avg,org.apache.kafka.consumer.connection.creation.total,org.apache.kafka.consumer.fetch.manager.fetch.total,org.apache.kafka.consumer.fetch.manager.bytes.consumed.rate,org.apache.kafka.consumer.request.total,org.apache.kafka.producer.bufferpool.wait.ratio,org.apache.kafka.producer.record.queue.time.avg,org.apache.kafka.producer.request.latency.avg,org.apache.kafka.producer.produce.throttle.time.avg,org.apache.kafka.producer.connection.creation.total,org.apache.kafka.producer.request.total,org.apache.kafka.producer.topic.byte.rate,org.apache.kafka.stream.client.state,org.apache.kafka.stream.thread.thread.state,org.apache.kafka.stream.state.size.all.mem.tables,org.apache.kafka.stream.state.estimate.num.keys,org.apache.kafka.stream.state.block.cache.usage,org.apache.kafka.stream.processor.node.record.e2e.latency.min,org.apache.kafka.stream.processor.node.record.e2e.latency.max,org.apache.kafka.stream.processor.node.record.e2e.latency.avg,org.apache.kafka.stream.thread.poll.ratio,org.apache.kafka.stream.thread.commit.ratio,org.apache.kafka.stream.thread.process.ratio,org.apache.kafka.stream.thread.punctuate.ratio,org.apache.kafka.stream.thread.process.records.avg"

kafka_broker_combined_properties: "{{kafka_broker_properties | confluent.platform.combine_properties}}"

kafka_broker_final_properties: "{{ kafka_broker_combined_properties | combine(kafka_broker_custom_properties) }}"

# Need complex jinja templating here, to be used by kafka broker listeners
plain_jaas_config: |-
  org.apache.kafka.common.security.plain.PlainLoginModule required username="{{sasl_plain_users_final.admin.principal}}" password="{{sasl_plain_users_final.admin.password}}" {% for user in sasl_plain_users_final|dict2items %} user_{{ user['value']['principal'] }}="{{ user['value']['password'] }}"{% endfor %};

# A set of client properties against the broker listener for kafka health checks
kafka_broker_default_client_properties: "{{ kafka_broker_listeners[kafka_broker_inter_broker_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            '', kafka_broker_pkcs12_truststore_path, kafka_broker_truststore_storepass, False, kafka_broker_pkcs12_keystore_path, kafka_broker_keystore_storepass, kafka_broker_keystore_keypass,
                            false, sasl_plain_users_final.admin.principal, sasl_plain_users_final.admin.password, sasl_scram_users_final.admin.principal, sasl_scram_users_final.admin.password, sasl_scram256_users_final.admin.principal, sasl_scram256_users_final.admin.password,
                            kerberos_kafka_broker_primary, kafka_broker_keytab_path, kafka_broker_kerberos_principal|default('kafka'),
                            false, kafka_broker_ldap_user, kafka_broker_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_broker_oauth_user, kafka_broker_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_broker_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_broker_third_party_oauth_client_assertion_config.kafka_broker)) }}"

kafka_broker_client_properties: "{{ kafka_broker_default_client_properties | combine(kafka_broker_custom_client_properties) }}"

#### Schema Registry Variables ####
schema_registry_service_name: confluent-schema-registry
schema_registry_default_user: cp-schema-registry
schema_registry_default_group: confluent
schema_registry_default_log_dir: /var/log/confluent/schema-registry
schema_registry:
  server_start_file: "{{ binary_base_path }}/bin/schema-registry-start"
  systemd_file: "{{systemd_base_dir}}/{{schema_registry_service_name}}.service"
  config_file: "{{ (config_base_path, schema_registry_config_prefix_path, 'schema-registry.properties') | path_join }}"
  systemd_override: /etc/systemd/system/{{schema_registry_service_name}}.service.d/override.conf
  log4j_file: "{{ (base_path, 'etc/schema-registry/log4j2.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, schema_registry_config_prefix_path, 'jaas.conf') | path_join }}"
  password_file: "{{ (config_base_path, schema_registry_config_prefix_path, 'password.properties') | path_join }}"

schema_registry_http_protocol: "{{ 'https' if schema_registry_ssl_enabled|bool else 'http' }}"

_schema_registry_url: "{{schema_registry_http_protocol}}://{{ (groups.get('schema_registry') if groups.get('schema_registry') else ['localhost']) | confluent.platform.resolve_and_format_hostnames(hostvars) | join(':' + schema_registry_listener_port|string + ',' + schema_registry_http_protocol + '://') }}:{{schema_registry_listener_port}}"
schema_registry_url: "{{ ccloud_schema_registry_url if ccloud_schema_registry_enabled else _schema_registry_url }}"

schema_registry_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost'])  | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[schema_registry_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[schema_registry_kafka_listener_name]['port']}}"

schema_registry_properties:
  defaults:
    enabled: true
    properties:
      debug: 'false'
      schema.registry.group.id: schema-registry
      kafkastore.topic: _schemas
      kafkastore.topic.replication.factor: "{{schema_registry_default_internal_replication_factor}}"
      listeners: "{{schema_registry_http_protocol}}://0.0.0.0:{{schema_registry_listener_port}}"
      host.name: "{{ hostvars[inventory_hostname]|confluent.platform.resolve_hostname }}"
      inter.instance.protocol: "{{schema_registry_http_protocol}}"
      kafkastore.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else schema_registry_bootstrap_servers }}"
      confluent.license.topic: _confluent-command
  ssl:
    enabled: "{{schema_registry_ssl_enabled}}"
    properties:
      security.protocol: SSL
      ssl.keystore.location: "{{schema_registry_keystore_path}}"
      ssl.keystore.password: "{{schema_registry_keystore_storepass}}"
      ssl.key.password: "{{schema_registry_keystore_keypass}}"
  principal_mapping_rules:
    enabled: "{{schema_registry_ssl_enabled}}"
    properties:
      auth.ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  truststore:
    enabled: "{{schema_registry_ssl_mutual_auth_enabled or mds_tls_enabled}}"
    properties:
      ssl.truststore.location: "{{schema_registry_truststore_path}}"
      ssl.truststore.password: "{{schema_registry_truststore_storepass}}"
  mtls:
    enabled: "{{ schema_registry_authentication_type == 'mtls' }}"
    properties:
      ssl.client.authentication: "{{ schema_registry_ssl_client_authentication | upper }}"
  basic:
    enabled: "{{ schema_registry_authentication_type == 'basic' }}"
    properties:
      authentication.method: BASIC
      authentication.realm: SchemaRegistry
      authentication.roles: "{{ schema_registry_basic_users_final | confluent.platform.get_roles | unique | join(',') }}"
  kafka_client:
    enabled: true
    properties: "{{ kafka_broker_listeners[schema_registry_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                    'kafkastore.', schema_registry_truststore_path, schema_registry_truststore_storepass, public_certificates_enabled, schema_registry_keystore_path, schema_registry_keystore_storepass, schema_registry_keystore_keypass,
                    false, sasl_plain_users_final.schema_registry.principal, sasl_plain_users_final.schema_registry.password, sasl_scram_users_final.schema_registry.principal, sasl_scram_users_final.schema_registry.password, sasl_scram256_users_final.schema_registry.principal, sasl_scram256_users_final.schema_registry.password,
                    kerberos_kafka_broker_primary, schema_registry_keytab_path, schema_registry_kerberos_principal|default('kafka'),
                    false, schema_registry_ldap_user, schema_registry_ldap_password, mds_bootstrap_server_urls, oauth_enabled, schema_registry_oauth_user, schema_registry_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, schema_registry_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(schema_registry_third_party_oauth_client_assertion_config.kafka)) }}"
  rbac:
    enabled: "{{ rbac_enabled }}"
    properties:
      confluent.schema.registry.authorizer.class: io.confluent.kafka.schemaregistry.security.authorizer.rbac.RbacAuthorizer
      confluent.schema.registry.auth.mechanism: JETTY_AUTH
      public.key.path: "{{rbac_enabled_public_pem_path}}"
      confluent.metadata.bootstrap.server.urls: "{{mds_bootstrap_server_urls}}"
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
  rbac_ldap:
    enabled: "{{ rbac_enabled and auth_mode == 'ldap' and (not schema_registry_mds_cert_auth_only|bool)}}"
    properties:
      confluent.metadata.http.auth.credentials.provider: BASIC
      confluent.metadata.basic.auth.user.info: "{{ schema_registry_ldap_user | default('MISSING') }}:{{ schema_registry_ldap_password | default('MISSING')}}"
  rbac_oauth:
    enabled: "{{rbac_enabled and oauth_enabled and (not schema_registry_mds_cert_auth_only|bool)}}"
    properties:
      confluent.metadata.http.auth.credentials.provider: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.id: "{{schema_registry_oauth_user}}"
      confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
  rbac_oauth_secret:
    enabled: "{{rbac_enabled and oauth_enabled and (not schema_registry_mds_cert_auth_only|bool) and (not schema_registry_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.secret: "{{schema_registry_oauth_password}}"
  rbac_oauth_scope:
    enabled: "{{rbac_enabled and oauth_enabled and oauth_groups_scope!='none' and (not schema_registry_mds_cert_auth_only|bool) and (not schema_registry_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.oauth.scope: "{{oauth_groups_scope}}"
  rbac_oauth_third_party_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path != 'none' and (not schema_registry_mds_cert_auth_only|bool)}}"
    properties:
      confluent.metadata.oauthbearer.assertion.file: "{{ schema_registry_third_party_oauth_client_assertion_config.mds }}"
  rbac_oauth_local_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not schema_registry_mds_cert_auth_only|bool)}}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.metadata.oauthbearer.assertion.claim.iss: "{{schema_registry_oauth_client_assertion_issuer}}"
      confluent.metadata.oauthbearer.assertion.claim.sub: "{{schema_registry_oauth_client_assertion_sub}}"
      confluent.metadata.oauthbearer.assertion.claim.aud: "{{schema_registry_oauth_client_assertion_audience}}"
      confluent.metadata.oauthbearer.assertion.private.key.file: "{{ schema_registry_oauth_client_assertion_private_key_file_dest_path }}"
  rbac_oauth_local_client_assertion_template_file:
    enabled: "{{rbac_enabled and oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not schema_registry_mds_cert_auth_only|bool) and schema_registry_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.template.file: "{{ schema_registry_oauth_client_assertion_template_file_dest_path }}"
  rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{rbac_enabled and oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not schema_registry_mds_cert_auth_only|bool) and schema_registry_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{schema_registry_oauth_client_assertion_private_key_passphrase}}"
  rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{rbac_enabled and oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not schema_registry_mds_cert_auth_only|bool) and schema_registry_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{schema_registry_oauth_client_assertion_jti_include|bool}}"
  rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{rbac_enabled and oauth_enabled and schema_registry_oauth_client_assertion_enabled and schema_registry_oauth_client_assertion_file_base_path == 'none' and (not schema_registry_mds_cert_auth_only|bool) and schema_registry_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{schema_registry_oauth_client_assertion_nbf_include}}"
  sr_oauth:
    enabled: "{{schema_registry_oauth_enabled}}"
    properties:
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      oauthbearer.jwks.endpoint.url: "{{oauth_jwks_uri}}"
      oauthbearer.sub.claim.name: "{{ oauth_sub_claim }}"
      oauthbearer.expected.issuer: "{{ oauth_issuer_url }}"
  sr_oauth_claim:
    enabled: "{{schema_registry_oauth_enabled and oauth_groups_claim!='none' }}"
    properties:
      oauthbearer.groups.claim.name: "{{ oauth_groups_claim }}"
  sr_oauth_audience:
    enabled: "{{schema_registry_oauth_enabled and oauth_expected_audience!='none' }}"
    properties:
      oauthbearer.expected.audience: "{{ oauth_expected_audience }}"
  sr_oauth_ssl:
    enabled: "{{schema_registry_oauth_enabled and idp_self_signed}}"
    properties:
      ssl.truststore.location: "{{schema_registry_truststore_path}}"
      ssl.truststore.password: "{{schema_registry_truststore_storepass}}"
  rbac_external_client:
    enabled: "{{ rbac_enabled and external_mds_enabled and mds_tls_enabled }}"
    properties:
      kafkastore.ssl.truststore.location: "{{schema_registry_truststore_path}}"
      kafkastore.ssl.truststore.password: "{{schema_registry_truststore_storepass}}"
  mds_client:
    enabled: "{{rbac_enabled and mds_tls_enabled }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{schema_registry_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{schema_registry_truststore_storepass}}"
      confluent.metadata.ssl.endpoint.identification.algorithm: HTTPS
  mds_mtls_client:
    enabled: "{{ rbac_enabled and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.ssl.keystore.location: "{{schema_registry_keystore_path}}"
      confluent.metadata.ssl.keystore.password: "{{schema_registry_keystore_storepass}}"
      confluent.metadata.ssl.key.password: "{{schema_registry_keystore_keypass}}"
      confluent.metadata.ssl.truststore.location: "{{schema_registry_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{schema_registry_truststore_storepass}}"
  oauth_mds_client:
    enabled: "{{ oauth_enabled and rbac_enabled and idp_self_signed and (not schema_registry_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{schema_registry_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{schema_registry_truststore_storepass}}"
  telemetry:
    enabled: "{{schema_registry_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter
  telemetry_proxy:
    enabled: "{{schema_registry_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{schema_registry_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{schema_registry_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"
  schema_exporter:
    enabled: "{{ schema_exporters is defined or enable_schema_exporter }}"
    properties:
      kafkastore.update.handlers: io.confluent.schema.exporter.storage.SchemaExporterUpdateHandler
  resource_extensions:
    enabled: "{{ rbac_enabled or (schema_exporters is defined or sr_switch_over_exporter_name is defined or enable_schema_exporter) or (schema_importers is defined or enable_schema_importer) or (unified_stream_manager.schema_registry_endpoint != 'none')}}"
    properties:
      schema.registry.resource.extension.class: "{{ rbac_enabled | confluent.platform.schema_registry_extension_classes(schema_exporters is defined or enable_schema_exporter, schema_importers is defined or enable_schema_importer, (unified_stream_manager.schema_registry_endpoint != 'none')) }}"
  schema_importer_and_exporter:
    enabled: "{{ schema_importers is defined or schema_exporters is defined or enable_schema_importer or enable_schema_exporter }}"
    properties:
      password.encoder.secret: "{{ password_encoder_secret }}"
  usm_sr:
    enabled: "{{unified_stream_manager.schema_registry_endpoint != 'none'}}" # enables the USM schema registry extension
    properties:
      usm.schema.registry.remote.endpoint: "{{ unified_stream_manager.schema_registry_endpoint }}"
      usm.schema.registry.remote.api.key: "{{ unified_stream_manager.basic_username }}"
      usm.schema.registry.remote.api.secret: "{{ unified_stream_manager.basic_password }}"
  usm_context_prefix:
    enabled: "{{unified_stream_manager.remote_context is defined and unified_stream_manager.remote_context != ''}}"
    properties:
      usm.schema.registry.remote.context.prefix: "{{ unified_stream_manager.remote_context | default('') }}"

schema_registry_combined_properties: "{{schema_registry_properties | confluent.platform.combine_properties}}"

schema_registry_final_properties: "{{ schema_registry_combined_properties | combine(schema_registry_custom_properties) }}"


#### Kafka Connect Variables ####
kafka_connect_default_service_name: confluent-kafka-connect
kafka_connect_default_config_filename: connect-distributed.properties
kafka_connect_default_user: cp-kafka-connect
kafka_connect_default_group: confluent
kafka_connect_default_log_dir: /var/log/kafka

kafka_connect:
  server_start_file: "{{ binary_base_path }}/bin/connect-distributed"
  systemd_file: "{{systemd_base_dir}}/{{kafka_connect_service_name}}.service"
  config_file: "{{ (config_base_path, kafka_connect_config_prefix_path, kafka_connect_config_filename) | path_join }}"
  systemd_override: /etc/systemd/system/{{kafka_connect_service_name}}.service.d/override.conf
  log4j_file: "{{ (base_path, 'etc/kafka/connect-log4j2.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, kafka_connect_config_prefix_path, 'connect-jaas.conf' ) | path_join }}"
  password_file: "{{ (config_base_path, kafka_connect_config_prefix_path, 'connect-password.properties') | path_join }}"

kafka_connect_http_protocol: "{{ 'https' if kafka_connect_ssl_enabled|bool else 'http' }}"

kafka_connect_rest_extension_classes:
  - "{% if rbac_enabled|bool %}io.confluent.connect.security.ConnectSecurityExtension{% endif %}"
  - "{% if kafka_connect_secret_registry_enabled|bool %}io.confluent.connect.secretregistry.ConnectSecretRegistryExtension{% endif %}"
  - "{% if kafka_connect_authentication_type == 'basic' %}org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension{% endif %}"

kafka_connect_final_rest_extension_classes: "{{(kafka_connect_rest_extension_classes|difference(['']) + kafka_connect_custom_rest_extension_classes) | unique}}"

kafka_connect_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_connect_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_connect_kafka_listener_name]['port']}}"
kafka_connect_producer_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_connect_producer_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_connect_producer_kafka_listener_name]['port']}}"
kafka_connect_consumer_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_connect_consumer_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_connect_consumer_kafka_listener_name]['port']}}"

kafka_connect_properties:
  defaults:
    enabled: true
    properties:
      rest.port: "{{kafka_connect_rest_port}}"
      config.storage.replication.factor: "{{ kafka_connect_default_internal_replication_factor }}"
      config.storage.topic: "{{kafka_connect_group_id}}-configs"
      group.id: "{{kafka_connect_group_id}}"
      internal.key.converter: org.apache.kafka.connect.json.JsonConverter
      internal.key.converter.schemas.enable: "false"
      internal.value.converter: org.apache.kafka.connect.json.JsonConverter
      internal.value.converter.schemas.enable: "false"
      offset.flush.interval.ms: 10000
      offset.storage.replication.factor: "{{ kafka_connect_default_internal_replication_factor }}"
      offset.storage.topic: "{{kafka_connect_group_id}}-offsets"
      status.storage.replication.factor: "{{ kafka_connect_default_internal_replication_factor }}"
      status.storage.topic: "{{kafka_connect_group_id}}-status"
      key.converter: io.confluent.connect.avro.AvroConverter
      value.converter: io.confluent.connect.avro.AvroConverter
      plugin.path: "{{(kafka_connect_plugins_path + [kafka_connect_confluent_hub_plugins_dest, kafka_connect_plugins_dest]) | unique | join(',')}}"
      connector.client.config.override.policy: All
      listeners: "{{kafka_connect_http_protocol}}://0.0.0.0:{{kafka_connect_rest_port}}"
      rest.advertised.listener: "{{kafka_connect_http_protocol}}"
      rest.advertised.host.name: "{{ hostvars[inventory_hostname]|confluent.platform.resolve_hostname }}"
      rest.advertised.port: "{{kafka_connect_rest_port}}"
      bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_connect_bootstrap_servers }}"
      producer.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_connect_producer_bootstrap_servers }}"
      consumer.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_connect_consumer_bootstrap_servers }}"
      confluent.license.topic: _confluent-command
  rest_classes:
    enabled: "{{ kafka_connect_final_rest_extension_classes|length > 0 }}"
    properties:
      rest.extension.classes: "{{ kafka_connect_final_rest_extension_classes | join(',') }}"
  ssl:
    enabled: "{{kafka_connect_ssl_enabled}}"
    properties:
      listeners.https.ssl.keystore.location: "{{kafka_connect_keystore_path}}"
      listeners.https.ssl.keystore.password: "{{kafka_connect_keystore_storepass}}"
      listeners.https.ssl.key.password: "{{kafka_connect_keystore_keypass}}"
      listeners.https.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      listeners.https.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
  mtls:
    enabled: "{{ kafka_connect_authentication_type == 'mtls' }}"
    properties:
      listeners.https.ssl.client.auth: "{{ kafka_connect_ssl_client_authentication }}"
  principal_mapping_rules:
    enabled: "{{ssl_enabled}}"
    properties:
      auth.ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  sr:
    enabled: "{{ 'schema_registry' in groups or ccloud_schema_registry_enabled|bool }}"
    properties:
      value.converter.schema.registry.url: "{{schema_registry_url}}"
      key.converter.schema.registry.url: "{{schema_registry_url}}"
  sr_ssl:
    enabled: "{{ 'schema_registry' in groups and schema_registry_ssl_enabled|bool and not ccloud_schema_registry_enabled }}"
    # This can also handle connect->sr communication over mtls
    properties:
      value.converter.schema.registry.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      value.converter.schema.registry.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
      value.converter.schema.registry.ssl.keystore.location: "{{kafka_connect_keystore_path}}"
      value.converter.schema.registry.ssl.keystore.password: "{{kafka_connect_keystore_storepass}}"
      value.converter.schema.registry.ssl.key.password: "{{kafka_connect_keystore_keypass}}"
      key.converter.schema.registry.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      key.converter.schema.registry.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
      key.converter.schema.registry.ssl.keystore.location: "{{kafka_connect_keystore_path}}"
      key.converter.schema.registry.ssl.keystore.password: "{{kafka_connect_keystore_storepass}}"
      key.converter.schema.registry.ssl.key.password: "{{kafka_connect_keystore_keypass}}"
  sr_basic:
    enabled: "{{ ('schema_registry' in groups and schema_registry_authentication_type == 'basic') or ccloud_schema_registry_enabled|bool }}"
    properties:
      value.converter.basic.auth.credentials.source: USER_INFO
      value.converter.schema.registry.basic.auth.user.info: "{{schema_registry_basic_users_final.admin.principal}}:{{schema_registry_basic_users_final.admin.password}}"
      key.converter.basic.auth.credentials.source: USER_INFO
      key.converter.schema.registry.basic.auth.user.info: "{{schema_registry_basic_users_final.admin.principal}}:{{schema_registry_basic_users_final.admin.password}}"
  kafka_client:
    enabled: true
    properties: "{{ kafka_broker_listeners[kafka_connect_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            '', kafka_connect_truststore_path, kafka_connect_truststore_storepass, public_certificates_enabled, kafka_connect_keystore_path, kafka_connect_keystore_storepass, kafka_connect_keystore_keypass,
                            false, sasl_plain_users_final.kafka_connect.principal, sasl_plain_users_final.kafka_connect.password, sasl_scram_users_final.kafka_connect.principal, sasl_scram_users_final.kafka_connect.password, sasl_scram256_users_final.kafka_connect.principal, sasl_scram256_users_final.kafka_connect.password,
                            kerberos_kafka_broker_primary, kafka_connect_keytab_path, kafka_connect_kerberos_principal|default('kafka'),
                            false, kafka_connect_ldap_user, kafka_connect_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_connect_oauth_user, kafka_connect_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_third_party_oauth_client_assertion_config.kafka)) }}"
  producer:
    enabled: true
    properties: "{{ kafka_broker_listeners[kafka_connect_producer_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            'producer.', kafka_connect_truststore_path, kafka_connect_truststore_storepass, public_certificates_enabled, kafka_connect_keystore_path, kafka_connect_keystore_storepass, kafka_connect_keystore_keypass,
                            false, sasl_plain_users_final.kafka_connect.principal, sasl_plain_users_final.kafka_connect.password, sasl_scram_users_final.kafka_connect.principal, sasl_scram_users_final.kafka_connect.password, sasl_scram256_users_final.kafka_connect.principal, sasl_scram256_users_final.kafka_connect.password,
                            kerberos_kafka_broker_primary, kafka_connect_keytab_path, kafka_connect_kerberos_principal|default('kafka'),
                            false, kafka_connect_ldap_user, kafka_connect_ldap_password, mds_bootstrap_server_urls, oauth_enabled and not ldap_with_oauth_enabled, kafka_connect_oauth_user, kafka_connect_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_third_party_oauth_client_assertion_config.producer)) }}"
  consumer:
    enabled: true
    properties: "{{ kafka_broker_listeners[kafka_connect_consumer_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            'consumer.', kafka_connect_truststore_path, kafka_connect_truststore_storepass, public_certificates_enabled, kafka_connect_keystore_path, kafka_connect_keystore_storepass, kafka_connect_keystore_keypass,
                            false, sasl_plain_users_final.kafka_connect.principal, sasl_plain_users_final.kafka_connect.password, sasl_scram_users_final.kafka_connect.principal, sasl_scram_users_final.kafka_connect.password, sasl_scram256_users_final.kafka_connect.principal, sasl_scram256_users_final.kafka_connect.password,
                            kerberos_kafka_broker_primary, kafka_connect_keytab_path, kafka_connect_kerberos_principal|default('kafka'),
                            false, kafka_connect_ldap_user, kafka_connect_ldap_password, mds_bootstrap_server_urls, oauth_enabled and not ldap_with_oauth_enabled, kafka_connect_oauth_user, kafka_connect_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_third_party_oauth_client_assertion_config.consumer)) }}"
  monitoring_interceptor:
    enabled: "{{ kafka_connect_monitoring_interceptors_enabled|bool }}"
    properties:
      confluent.monitoring.interceptor.topic: _confluent-monitoring
      producer.confluent.monitoring.interceptor.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_connect_bootstrap_servers }}"
      consumer.confluent.monitoring.interceptor.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_connect_bootstrap_servers }}"
  producer_monitoring_interceptor_client:
    enabled: "{{ kafka_connect_monitoring_interceptors_enabled|bool }}"
    properties: "{{ kafka_broker_listeners[kafka_connect_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            'producer.confluent.monitoring.interceptor.', kafka_connect_truststore_path, kafka_connect_truststore_storepass, public_certificates_enabled, kafka_connect_keystore_path, kafka_connect_keystore_storepass, kafka_connect_keystore_keypass,
                            false, sasl_plain_users_final.kafka_connect.principal, sasl_plain_users_final.kafka_connect.password, sasl_scram_users_final.kafka_connect.principal, sasl_scram_users_final.kafka_connect.password, sasl_scram256_users_final.kafka_connect.principal, sasl_scram256_users_final.kafka_connect.password,
                            kerberos_kafka_broker_primary, kafka_connect_keytab_path, kafka_connect_kerberos_principal|default('kafka'),
                            false, kafka_connect_ldap_user, kafka_connect_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_connect_oauth_user, kafka_connect_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_third_party_oauth_client_assertion_config.producer_monitoring_interceptor)) }}"
  consumer_monitoring_interceptor_client:
    enabled: "{{ kafka_connect_monitoring_interceptors_enabled|bool }}"
    properties: "{{ kafka_broker_listeners[kafka_connect_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            'consumer.confluent.monitoring.interceptor.', kafka_connect_truststore_path, kafka_connect_truststore_storepass, public_certificates_enabled, kafka_connect_keystore_path, kafka_connect_keystore_storepass, kafka_connect_keystore_keypass,
                            false, sasl_plain_users_final.kafka_connect.principal, sasl_plain_users_final.kafka_connect.password, sasl_scram_users_final.kafka_connect.principal, sasl_scram_users_final.kafka_connect.password, sasl_scram256_users_final.kafka_connect.principal, sasl_scram256_users_final.kafka_connect.password,
                            kerberos_kafka_broker_primary, kafka_connect_keytab_path, kafka_connect_kerberos_principal|default('kafka'),
                            false, kafka_connect_ldap_user, kafka_connect_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_connect_oauth_user, kafka_connect_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_third_party_oauth_client_assertion_config.consumer_monitoring_interceptor)) }}"
  rbac:
    enabled: "{{rbac_enabled}}"
    properties:
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      public.key.path: "{{rbac_enabled_public_pem_path}}"
      confluent.metadata.bootstrap.server.urls: "{{mds_bootstrap_server_urls}}"
  rbac_ldap:
    enabled: "{{ rbac_enabled and auth_mode == 'ldap' and (not kafka_connect_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.http.auth.credentials.provider: BASIC
      confluent.metadata.basic.auth.user.info: "{{kafka_connect_ldap_user| default('MISSING')}}:{{kafka_connect_ldap_password| default('MISSING')}}"
  rbac_oauth:
    enabled: "{{ rbac_enabled and oauth_enabled and (not kafka_connect_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.http.auth.credentials.provider: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.id: "{{kafka_connect_oauth_user}}"
      confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
  rbac_oauth_secret:
    enabled: "{{ rbac_enabled and oauth_enabled and (not kafka_connect_mds_cert_auth_only) and (not kafka_connect_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.secret: "{{kafka_connect_oauth_password}}"
  rbac_oauth_scope:
    enabled: "{{rbac_enabled and oauth_enabled and oauth_groups_scope!='none' and (not kafka_connect_mds_cert_auth_only) and (not kafka_connect_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.oauth.scope: "{{oauth_groups_scope}}"
  rbac_oauth_third_party_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_connect_oauth_client_assertion_enabled and kafka_connect_oauth_client_assertion_file_base_path !='none' and (not kafka_connect_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.assertion.file: "{{ kafka_connect_third_party_oauth_client_assertion_config.mds }}"
  rbac_oauth_local_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_connect_oauth_client_assertion_enabled and kafka_connect_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.metadata.oauthbearer.assertion.claim.iss: "{{kafka_connect_oauth_client_assertion_issuer}}"
      confluent.metadata.oauthbearer.assertion.claim.sub: "{{kafka_connect_oauth_client_assertion_sub}}"
      confluent.metadata.oauthbearer.assertion.claim.aud: "{{kafka_connect_oauth_client_assertion_audience}}"
      confluent.metadata.oauthbearer.assertion.private.key.file: "{{ kafka_connect_oauth_client_assertion_private_key_file_dest_path }}"
  rbac_oauth_local_client_assertion_template_file:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_connect_oauth_client_assertion_enabled and kafka_connect_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_mds_cert_auth_only) and kafka_connect_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.template.file: "{{ kafka_connect_oauth_client_assertion_template_file_dest_path }}"
  rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_connect_oauth_client_assertion_enabled and kafka_connect_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_mds_cert_auth_only) and kafka_connect_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{kafka_connect_oauth_client_assertion_private_key_passphrase}}"
  rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_connect_oauth_client_assertion_enabled and kafka_connect_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_mds_cert_auth_only) and kafka_connect_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{kafka_connect_oauth_client_assertion_jti_include}}"
  rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_connect_oauth_client_assertion_enabled and kafka_connect_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_mds_cert_auth_only) and kafka_connect_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{kafka_connect_oauth_client_assertion_nbf_include}}"
  kafka_connect_oauth:
    enabled: "{{kafka_connect_oauth_enabled}}"
    properties:
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      oauthbearer.jwks.endpoint.url: "{{oauth_jwks_uri}}"
      oauthbearer.sub.claim.name: "{{ oauth_sub_claim }}"
      oauthbearer.expected.issuer: "{{ oauth_issuer_url }}"
  kafka_connect_oauth_claim:
    enabled: "{{kafka_connect_oauth_enabled and oauth_groups_claim!='none' }}"
    properties:
      oauthbearer.groups.claim.name: "{{ oauth_groups_claim }}"
  kafka_connect_oauth_audience:
    enabled: "{{kafka_connect_oauth_enabled and oauth_expected_audience!='none' }}"
    properties:
      oauthbearer.expected.audience: "{{ oauth_expected_audience }}"
  kafka_connect_oauth_ssl:
    enabled: "{{kafka_connect_oauth_enabled and idp_self_signed}}"
    properties:
      ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
  rbac_external_client:
    enabled: "{{ rbac_enabled and external_mds_enabled and mds_tls_enabled }}"
    properties:
      consumer.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      consumer.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
      producer.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      producer.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
      ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
      config.providers.secret.param.kafkastore.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      config.providers.secret.param.kafkastore.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
  mds_client:
    enabled: "{{rbac_enabled and mds_tls_enabled }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
      confluent.metadata.ssl.endpoint.identification.algorithm: HTTPS
  mds_mtls_client:
    enabled: "{{ rbac_enabled and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.ssl.keystore.location: "{{kafka_connect_keystore_path}}"
      confluent.metadata.ssl.keystore.password: "{{kafka_connect_keystore_storepass}}"
      confluent.metadata.ssl.key.password: "{{kafka_connect_keystore_keypass}}"
      confluent.metadata.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
  oauth_mds_client:
    enabled: "{{ oauth_enabled and rbac_enabled and idp_self_signed }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{kafka_connect_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_connect_truststore_storepass}}"
  secret_registry:
    enabled: "{{kafka_connect_secret_registry_enabled}}"
    properties:
      config.providers: secret
      config.providers.secret.class: io.confluent.connect.secretregistry.rbac.config.provider.InternalSecretConfigProvider
      config.providers.secret.param.master.encryption.key: "{{kafka_connect_secret_registry_key}}"
      config.providers.secret.param.kafkastore.topic: _confluent-secrets
      config.providers.secret.param.kafkastore.topic.replication.factor: "{{kafka_connect_secret_registry_default_replication_factor}}"
      config.providers.secret.param.secret.registry.group.id: secret-registry
      config.providers.secret.param.kafkastore.bootstrap.servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_connect_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_connect_kafka_listener_name]['port']}}"
  secret_registry_client:
    enabled: true
    properties: "{{ kafka_broker_listeners[kafka_connect_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                    'config.providers.secret.param.kafkastore.', kafka_connect_truststore_path, kafka_connect_truststore_storepass, public_certificates_enabled, kafka_connect_keystore_path, kafka_connect_keystore_storepass, kafka_connect_keystore_keypass,
                    false, sasl_plain_users_final.kafka_connect.principal, sasl_plain_users_final.kafka_connect.password, sasl_scram_users_final.kafka_connect.principal, sasl_scram_users_final.kafka_connect.password, sasl_scram256_users_final.kafka_connect.principal, sasl_scram256_users_final.kafka_connect.password,
                    kerberos_kafka_broker_primary, kafka_connect_keytab_path, kafka_connect_kerberos_principal|default('kafka'),
                    false, kafka_connect_ldap_user, kafka_connect_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_connect_oauth_user, kafka_connect_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_third_party_oauth_client_assertion_config.secret_registry)) }}"
  telemetry:
    enabled: "{{kafka_connect_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter
  telemetry_proxy:
    enabled: "{{kafka_connect_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{kafka_connect_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{kafka_connect_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"
  usm_agent_telemetry:
    enabled: "{{ 'usm_agent' in groups }}"
    properties:
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter
      confluent.telemetry.exporter._usm.enabled: true
      confluent.telemetry.exporter._usm.type: http
      confluent.telemetry.exporter._usm.client.base.url: "{{ usm_agent_url }}"
      confluent.connect.metadata.events.client.base.url: "{{ usm_agent_url }}"
      confluent.connect.metadata.events.enabled: true
      confluent.connect.metadata.events.emitter.type: cloud.events.http
  usm_agent_telemetry_auth_basic:
    enabled: "{{ 'usm_agent' in groups and usm_agent_basic_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.api.key: "{{ kafka_connect_usm_agent_username }}"
      confluent.telemetry.exporter._usm.api.secret: "{{ kafka_connect_usm_agent_password }}"
      confluent.connect.metadata.events.api.key: "{{ kafka_connect_usm_agent_username }}"
      confluent.connect.metadata.events.api.secret: "{{ kafka_connect_usm_agent_password }}"
  usm_agent_telemetry_auth_dummy:
    enabled: "{{ 'usm_agent' in groups and not usm_agent_basic_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.api.key: dummy
      confluent.telemetry.exporter._usm.api.secret: dummy
      confluent.connect.metadata.events.api.key: dummy
      confluent.connect.metadata.events.api.secret: dummy
  usm_agent_telemetry_ssl:
    enabled: "{{ 'usm_agent' in groups and usm_agent_ssl_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.https.ssl.protocol: TLSv1.2
      confluent.telemetry.exporter._usm.https.ssl.truststore.location: "{{ kafka_connect_truststore_path }}"
      confluent.telemetry.exporter._usm.https.ssl.truststore.password: "{{ kafka_connect_truststore_storepass }}"
      confluent.connect.metadata.events.ssl.protocol: TLSv1.2
      confluent.connect.metadata.events.ssl.truststore.location: "{{ kafka_connect_truststore_path }}"
      confluent.connect.metadata.events.ssl.truststore.password: "{{ kafka_connect_truststore_storepass }}"
  usm_agent_telemetry_mtls:
    enabled: "{{ 'usm_agent' in groups and usm_agent_ssl_mutual_auth_enabled }}"
    properties:
      confluent.telemetry.exporter._usm.https.ssl.keystore.location: "{{ kafka_connect_keystore_path }}"
      confluent.telemetry.exporter._usm.https.ssl.keystore.password: "{{ kafka_connect_keystore_storepass }}"
      confluent.telemetry.exporter._usm.https.ssl.key.password: "{{ kafka_connect_keystore_keypass }}"
      confluent.connect.metadata.events.ssl.keystore.location: "{{ kafka_connect_keystore_path }}"
      confluent.connect.metadata.events.ssl.keystore.password: "{{ kafka_connect_keystore_storepass }}"
      confluent.connect.metadata.events.ssl.key.password: "{{ kafka_connect_keystore_keypass }}"

kafka_connect_combined_properties: "{{kafka_connect_properties | confluent.platform.combine_properties}}"

kafka_connect_final_properties: "{{ kafka_connect_combined_properties | combine(kafka_connect_custom_properties) }}"


#### KSQLDB Variables ####
ksql_service_name: "{{(confluent_package_version is version('5.5.0', '>=')) | ternary('confluent-ksqldb' , 'confluent-ksql')}}"
ksql_main_package: "{{(confluent_package_version is version('5.5.0', '>=')) | ternary('confluent-ksqldb' , 'confluent-ksql')}}"
ksql_default_user: cp-ksql
ksql_default_group: confluent
ksql_default_log_dir: /var/log/confluent/ksql
ksql:
  server_start_file: "{{ binary_base_path }}/bin/ksql-server-start"
  config_file: "{{ (config_base_path, ksql_config_prefix_path, 'ksql-server.properties') | path_join }}"
  systemd_file: "{{ (systemd_base_dir, (ksql_service_name, '.service') | join) | path_join }}"
  systemd_override: "{{ ('/etc/systemd/system', (ksql_service_name, '.service.d') | join , 'override.conf') | path_join }}"
  log4j_file: "{{ (base_path, 'etc/ksqldb/ksqldb-log4j2.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, ('etc/ksql', ('db' if (confluent_package_version is version('5.5.0', '>=')) else '')) | join, 'jaas.conf' ) | path_join }}"
  password_file: "{{ (config_base_path, ('etc/ksql', ('db' if (confluent_package_version is version('5.5.0', '>=')) else '')) | join, 'password.properties') | path_join }}"

ksql_http_protocol: "{{ 'https' if ksql_ssl_enabled|bool else 'http' }}"

ksql_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[ksql_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[ksql_kafka_listener_name]['port']}}"

ksql_properties:
  defaults:
    enabled: true
    properties:
      application.id: ksql-server
      ksql.service.id: "{{ ksql_service_id }}"
      ksql.internal.topic.replicas: "{{ ksql_default_internal_replication_factor }}"
      ksql.streams.replication.factor: "{{ ksql_default_internal_replication_factor }}"
      ksql.streams.state.dir: /var/lib/kafka-streams
      ksql.streams.num.standby.replicas: 1
      ksql.streams.producer.delivery.timeout.ms: 2147483647
      ksql.streams.producer.max.block.ms: 9223372036854775807
      listeners: "{{ksql_http_protocol}}://0.0.0.0:{{ksql_listener_port}}"
      bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else ksql_bootstrap_servers }}"
      security.protocol: "{{kafka_broker_listeners[ksql_kafka_listener_name] | confluent.platform.kafka_protocol_defaults(ssl_enabled, sasl_protocol) }}"
  ssl:
    # KSQL SSL properties shared with Kafka Broker
    enabled: "{{ ksql_ssl_enabled|bool or ( kafka_broker_listeners[ksql_kafka_listener_name]['ssl_enabled']|default(ssl_enabled)|bool and not ccloud_kafka_enabled|bool ) }}"
    properties:
      ssl.truststore.location: "{{ksql_truststore_path}}"
      ssl.truststore.password: "{{ksql_truststore_storepass}}"
      ssl.keystore.location: "{{ksql_keystore_path}}"
      ssl.keystore.password: "{{ksql_keystore_storepass}}"
      ssl.key.password: "{{ksql_keystore_keypass}}"
  mtls:
    enabled: "{{ ksql_authentication_type == 'mtls' }}"
    properties:
      ssl.client.auth: 'true'
  principal_mapping_rules:
    enabled: "{{ssl_enabled}}"
    properties:
      ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  basic:
    enabled: "{{ ksql_authentication_type == 'basic' }}"
    properties:
      authentication.method: BASIC
      authentication.realm: KsqlServer
      authentication.roles: "{{ ksql_basic_users | confluent.platform.get_roles | unique | join(',') }}"
  kafka_sasl_plain:
    enabled: "{{ (kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'PLAIN' }}"
    properties:
      sasl.mechanism: PLAIN
      sasl.jaas.config: |-
        org.apache.kafka.common.security.plain.PlainLoginModule required username="{{sasl_plain_users_final.ksql.principal}}" password="{{sasl_plain_users_final.ksql.password}}";
  kafka_sasl_scram:
    enabled: "{{ (kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'SCRAM-SHA-512' }}"
    properties:
      sasl.mechanism: SCRAM-SHA-512
      sasl.jaas.config: |-
        org.apache.kafka.common.security.scram.ScramLoginModule required username="{{sasl_scram_users_final.ksql.principal}}" password="{{sasl_scram_users_final.ksql.password}}";
  kafka_sasl_scram256:
    enabled: "{{ (kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'SCRAM-SHA-256' }}"
    properties:
      sasl.mechanism: SCRAM-SHA-256
      sasl.jaas.config: |-
        org.apache.kafka.common.security.scram.ScramLoginModule required username="{{sasl_scram256_users_final.ksql.principal}}" password="{{sasl_scram256_users_final.ksql.password}}";
  kafka_sasl_gssapi:
    enabled: "{{ (kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'GSSAPI' }}"
    properties:
      sasl.mechanism: GSSAPI
      sasl.kerberos.service.name: "{{kerberos_kafka_broker_primary}}"
      sasl.jaas.config: |-
        com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="{{ksql_keytab_path}}" principal="{{ksql_kerberos_principal | default('ksql')}}";
  kafka_sasl_ldap:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and (auth_mode == 'ldap') }}"
    properties:
      sasl.mechanism: OAUTHBEARER
      sasl.login.callback.handler.class: io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
      sasl.jaas.config: |-
        org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required username="{{ksql_ldap_user | default('ksql')}}" password="{{ksql_ldap_password | default('pass')}}" metadataServerUrls="{{mds_bootstrap_server_urls}}";
  kafka_sasl_oauth:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled }}"
    properties:
      sasl.mechanism: OAUTHBEARER
      sasl.login.callback.handler.class: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
      sasl.oauthbearer.token.endpoint.url: "{{oauth_token_uri}}"
  kafka_sasl_oauth_secret_jaas:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and (not ksql_oauth_client_assertion_enabled) }}"
    properties:
      sasl.jaas.config: |-
        org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{{ksql_oauth_user}}" clientSecret="{{ksql_oauth_password}}" {% if oauth_groups_scope!='none' %}scope="{{oauth_groups_scope}}"{% endif %} {% if idp_self_signed %}ssl.truststore.location="{{ksql_truststore_path}}" ssl.truststore.password="{{ksql_truststore_storepass}}"{% endif %};
  kafka_sasl_oauth_third_party_client_assertion:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path !='none' }}"
    properties:
      sasl.oauthbearer.assertion.file: "{{ ksql_third_party_oauth_client_assertion_config.kafka }}"
      sasl.jaas.config: |-
        org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{{ksql_oauth_user}}" {% if oauth_groups_scope!='none' %}scope="{{oauth_groups_scope}}"{% endif %} {% if idp_self_signed %}ssl.truststore.location="{{ksql_truststore_path}}" ssl.truststore.password="{{ksql_truststore_storepass}}"{% endif %};
  kafka_sasl_oauth_local_client_assertion:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' }}"
    properties:
      sasl.oauthbearer.assertion.claim.iss: "{{ksql_oauth_client_assertion_issuer}}"
      sasl.oauthbearer.assertion.claim.sub: "{{ksql_oauth_client_assertion_sub}}"
      sasl.oauthbearer.assertion.claim.aud: "{{ksql_oauth_client_assertion_audience}}"
      sasl.oauthbearer.assertion.private.key.file: "{{ ksql_oauth_client_assertion_private_key_file_dest_path }}"
      sasl.jaas.config: |-
        org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{{ksql_oauth_user}}" {% if oauth_groups_scope!='none' %}scope="{{oauth_groups_scope}}"{% endif %} {% if idp_self_signed %}ssl.truststore.location="{{ksql_truststore_path}}" ssl.truststore.password="{{ksql_truststore_storepass}}"{% endif %};
  kafka_sasl_oauth_client_assertion_template_file:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none'  and ksql_oauth_client_assertion_template_file != 'none'}}"
    properties:
      sasl.oauthbearer.assertion.template.file: "{{ ksql_oauth_client_assertion_template_file_dest_path }}"
  kafka_sasl_oauth_client_assertion_passphrase:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none'  and ksql_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      sasl.oauthbearer.assertion.private.key.passphrase: "{{ksql_oauth_client_assertion_private_key_passphrase}}"
  kafka_sasl_oauth_client_assertion_jti_include:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none'  and ksql_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      sasl.oauthbearer.assertion.claim.jti.include: "{{ksql_oauth_client_assertion_jti_include}}"
  kafka_sasl_oauth_client_assertion_nbf_include:
    enabled: "{{ ((kafka_broker_listeners[ksql_kafka_listener_name]['sasl_protocol'] | default(sasl_protocol) | confluent.platform.normalize_sasl_protocol)[0] == 'OAUTHBEARER' ) and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none'  and ksql_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      sasl.oauthbearer.assertion.claim.nbf.include: "{{ksql_oauth_client_assertion_nbf_include}}"
  sr:
    enabled: "{{ 'schema_registry' in groups }}"
    properties:
      ksql.schema.registry.url: "{{schema_registry_url}}"
  sr_ssl:
    enabled: "{{ 'schema_registry' in groups and schema_registry_ssl_enabled|bool and not ccloud_schema_registry_enabled|bool }}"
    properties:
      ksql.schema.registry.ssl.truststore.location: "{{ksql_truststore_path}}"
      ksql.schema.registry.ssl.truststore.password: "{{ksql_truststore_storepass}}"
      ksql.schema.registry.ssl.keystore.location: "{{ksql_keystore_path}}"
      ksql.schema.registry.ssl.keystore.password: "{{ksql_keystore_storepass}}"
      ksql.schema.registry.ssl.key.password: "{{ksql_keystore_keypass}}"
  sr_basic:
    enabled: "{{ ('schema_registry' in groups and schema_registry_authentication_type == 'basic') or ccloud_schema_registry_enabled|bool }}"
    properties:
      ksql.schema.registry.basic.auth.credentials.source: USER_INFO
      ksql.schema.registry.basic.auth.user.info: "{{schema_registry_basic_users_final.admin.principal}}:{{schema_registry_basic_users_final.admin.password}}"
  sr_rbac:
    enabled: "{{ 'schema_registry' in groups and rbac_enabled|bool and ((not schema_registry_oauth_enabled) or ldap_with_oauth_enabled)}}"
    properties:
      ksql.schema.registry.basic.auth.credentials.source: USER_INFO
      ksql.schema.registry.basic.auth.user.info: "{{ ksql_ldap_user | default('ksql') }}:{{ ksql_ldap_password | default('pass') }}"
  sr_oauth:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and (not ldap_with_oauth_enabled)}}"
    properties:
      ksql.schema.registry.bearer.auth.client.id: "{{ksql_oauth_user}}"
      ksql.schema.registry.bearer.auth.issuer.endpoint.url: "{{oauth_token_uri}}"
  sr_oauth_secret:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and (not ldap_with_oauth_enabled) and (not ksql_oauth_client_assertion_enabled) }}"
    properties:
      ksql.schema.registry.bearer.auth.credentials.source: OAUTHBEARER
      ksql.schema.registry.bearer.auth.client.secret: "{{ksql_oauth_password}}"
  sr_oauth_scope:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and (not ldap_with_oauth_enabled) and oauth_groups_scope!='none' and (not ksql_oauth_client_assertion_enabled) }}"
    properties:
      ksql.schema.registry.bearer.auth.scope: "{{oauth_groups_scope}}"
  sr_oauth_ssl:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and (not ldap_with_oauth_enabled) and idp_self_signed }}"
    properties:
      ksql.schema.registry.ssl.truststore.location: "{{ksql_truststore_path}}"
      ksql.schema.registry.ssl.truststore.password: "{{ksql_truststore_storepass}}"
  sr_oauth_third_party_client_assertion:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path !='none' and (not ldap_with_oauth_enabled)}}"
    properties:
      ksql.schema.registry.bearer.auth.credentials.source: OAUTHBEARER_CLIENTASSERTION
      ksql.schema.registry.bearer.assertion.file: "{{ksql_third_party_oauth_client_assertion_config.schema_registry}}"
  sr_oauth_local_client_assertion:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and ksql_oauth_client_assertion_file_base_path =='none' and ksql_oauth_client_assertion_enabled and (not ldap_with_oauth_enabled)}}"
    properties:
      ksql.schema.registry.bearer.auth.credentials.source: OAUTHBEARER_CLIENTASSERTION
      ksql.schema.registry.bearer.assertion.claim.iss: "{{ksql_oauth_client_assertion_issuer}}"
      ksql.schema.registry.bearer.assertion.claim.sub: "{{ksql_oauth_client_assertion_sub}}"
      ksql.schema.registry.bearer.assertion.claim.aud: "{{ksql_oauth_client_assertion_audience}}"
      ksql.schema.registry.bearer.assertion.private.key.file: "{{ ksql_oauth_client_assertion_private_key_file_dest_path }}"
  sr_oauth_local_client_assertion_template_file:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ldap_with_oauth_enabled) and ksql_oauth_client_assertion_template_file != 'none'}}"
    properties:
      ksql.schema.registry.bearer.assertion.template.file: "{{ ksql_oauth_client_assertion_template_file_dest_path }}"
  sr_oauth_local_client_assertion_passphrase:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ldap_with_oauth_enabled) and ksql_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      ksql.schema.registry.bearer.assertion.private.key.passphrase: "{{ksql_oauth_client_assertion_private_key_passphrase}}"
  sr_oauth_local_client_assertion_jti_include:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ldap_with_oauth_enabled) and ksql_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      ksql.schema.registry.bearer.assertion.claim.jti.include: "{{ksql_oauth_client_assertion_jti_include}}"
  sr_oauth_local_client_assertion_nbf_include:
    enabled: "{{ 'schema_registry' in groups and schema_registry_oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ldap_with_oauth_enabled) and ksql_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      ksql.schema.registry.bearer.assertion.claim.nbf.include: "{{ksql_oauth_client_assertion_nbf_include}}"
  monitoring_interceptor:
    enabled: "{{ ksql_monitoring_interceptors_enabled|bool }}"
    properties:
      confluent.monitoring.interceptor.topic: _confluent-monitoring
      confluent.monitoring.interceptor.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else ksql_bootstrap_servers }}"
  monitoring_interceptor_client:
    enabled: "{{ ksql_monitoring_interceptors_enabled|bool }}"
    properties: "{{ kafka_broker_listeners[ksql_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            'confluent.monitoring.interceptor.', ksql_truststore_path, ksql_truststore_storepass, public_certificates_enabled, ksql_keystore_path, ksql_keystore_storepass, ksql_keystore_keypass,
                            false, sasl_plain_users_final.ksql.principal, sasl_plain_users_final.ksql.password, sasl_scram_users_final.ksql.principal, sasl_scram_users_final.ksql.password, sasl_scram256_users_final.ksql.principal, sasl_scram256_users_final.ksql.password,
                            kerberos_kafka_broker_primary, ksql_keytab_path, ksql_kerberos_principal|default('ksql'),
                            false, ksql_ldap_user, ksql_ldap_password, mds_bootstrap_server_urls, oauth_enabled and (not ldap_with_oauth_enabled), ksql_oauth_user, ksql_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, ksql_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(ksql_third_party_oauth_client_assertion_config.monitoring_interceptor)) }}"
  rbac:
    enabled: "{{rbac_enabled}}"
    properties:
      public.key.path: "{{rbac_enabled_public_pem_path}}"
      confluent.metadata.bootstrap.server.urls: "{{mds_bootstrap_server_urls}}"
  rbac_ldap:
    enabled: "{{ rbac_enabled and auth_mode == 'ldap' and (not ksql_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.basic.auth.user.info: "{{ ksql_ldap_user | default('ksql') }}:{{ ksql_ldap_password | default('pass') }}"
      confluent.metadata.http.auth.credentials.provider: BASIC
      ksql.security.extension.class: io.confluent.ksql.security.KsqlConfluentSecurityExtension
      ksql.authentication.plugin.class: io.confluent.ksql.security.VertxAuthenticationHandlerPlugin
  rbac_oauth:
    enabled: "{{ rbac_enabled and oauth_enabled and (not ksql_mds_cert_auth_only ) }}"
    properties:
      confluent.metadata.oauthbearer.login.client.id: "{{ksql_oauth_user}}"
      confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
      confluent.metadata.http.auth.credentials.provider: OAUTHBEARER
      ksql.security.extension.class: io.confluent.ksql.security.KsqlConfluentSecurityExtension
      ksql.authentication.plugin.class: io.confluent.ksql.security.VertxAuthenticationHandlerPlugin
  rbac_oauth_secret:
    enabled: "{{ rbac_enabled and oauth_enabled and (not ksql_mds_cert_auth_only ) and (not ksql_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.secret: "{{ksql_oauth_password}}"
  rbac_oauth_scope:
    enabled: "{{ rbac_enabled and oauth_enabled and oauth_groups_scope!='none' and (not ksql_mds_cert_auth_only) and (not ksql_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.oauth.scope: "{{ oauth_groups_scope }}"
  rbac_oauth_third_party_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path !='none' and (not ksql_mds_cert_auth_only ) }}"
    properties:
      confluent.metadata.oauthbearer.assertion.file: "{{ ksql_third_party_oauth_client_assertion_config.mds }}"
  rbac_oauth_local_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ksql_mds_cert_auth_only ) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.metadata.oauthbearer.assertion.claim.iss: "{{ksql_oauth_client_assertion_issuer}}"
      confluent.metadata.oauthbearer.assertion.claim.sub: "{{ksql_oauth_client_assertion_sub}}"
      confluent.metadata.oauthbearer.assertion.claim.aud: "{{ksql_oauth_client_assertion_audience}}"
      confluent.metadata.oauthbearer.assertion.private.key.file: "{{ ksql_oauth_client_assertion_private_key_file_dest_path }}"
  rbac_oauth_local_client_assertion_template_file:
    enabled: "{{rbac_enabled and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ksql_mds_cert_auth_only ) and ksql_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.template.file: "{{ ksql_oauth_client_assertion_template_file_dest_path }}"
  rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{rbac_enabled and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ksql_mds_cert_auth_only ) and ksql_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{ksql_oauth_client_assertion_private_key_passphrase}}"
  rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{rbac_enabled and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ksql_mds_cert_auth_only ) and ksql_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{ksql_oauth_client_assertion_jti_include}}"
  rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{rbac_enabled and oauth_enabled and ksql_oauth_client_assertion_enabled and ksql_oauth_client_assertion_file_base_path =='none' and (not ksql_mds_cert_auth_only ) and ksql_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{ksql_oauth_client_assertion_nbf_include}}"
  ksql_oauth:
    enabled: "{{ksql_oauth_enabled}}"
    properties:
      ksql.authentication.plugin.class: io.confluent.ksql.security.VertxAuthenticationHandlerPlugin
      oauthbearer.jwks.endpoint.url: "{{oauth_jwks_uri}}"
      oauthbearer.sub.claim.name: "{{ oauth_sub_claim }}"
      oauthbearer.expected.issuer: "{{ oauth_issuer_url }}"
  ksql_oauth_claim:
    enabled: "{{ksql_oauth_enabled and oauth_groups_scope!='none' }}"
    properties:
      oauthbearer.groups.claim.name: "{{ oauth_groups_claim }}"
  ksql_oauth_audience:
    enabled: "{{ksql_oauth_enabled and oauth_expected_audience!='none' }}"
    properties:
      oauthbearer.expected.audience: "{{ oauth_expected_audience }}"
  ksql_oauth_ssl:
    enabled: "{{ksql_oauth_enabled and idp_self_signed}}"
    properties:
      ssl.truststore.location: "{{ksql_truststore_path}}"
      ssl.truststore.password: "{{ksql_truststore_storepass}}"
  rbac_external_client:
    enabled: "{{rbac_enabled and external_mds_enabled and mds_tls_enabled }}"
    properties:
      ssl.truststore.location: "{{ksql_truststore_path}}"
      ssl.truststore.password: "{{ksql_truststore_storepass}}"
  mds_client:
    enabled: "{{rbac_enabled and mds_tls_enabled }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{ksql_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{ksql_truststore_storepass}}"
      confluent.metadata.ssl.endpoint.identification.algorithm: HTTPS
  mds_mtls_client:
    enabled: "{{ rbac_enabled and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.ssl.keystore.location: "{{ksql_keystore_path}}"
      confluent.metadata.ssl.keystore.password: "{{ksql_keystore_keypass}}"
      confluent.metadata.ssl.key.password: "{{ksql_keystore_keypass}}"
      confluent.metadata.ssl.truststore.location: "{{ksql_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{ksql_truststore_storepass}}"
  telemetry:
    enabled: "{{ksql_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter
  telemetry_proxy:
    enabled: "{{ksql_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{ksql_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{ksql_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"
  log_streaming:
    enabled: "{{ksql_log_streaming_enabled}}"
    properties:
      ksql.logging.processing.topic.auto.create: "{{ksql_log_streaming_enabled | lower}}"
      ksql.logging.processing.topic.name: "{{ksql_service_id}}{{ksql_processing_log}}"
      ksql.logging.processing.topic.replication.factor: "{{ ksql_default_internal_replication_factor }}"

ksql_combined_properties: "{{ksql_properties | confluent.platform.combine_properties}}"

ksql_final_properties: "{{ ksql_combined_properties | combine(ksql_custom_properties) }}"


#### Kafka Rest Variables ####
kafka_rest_service_name: confluent-kafka-rest
kafka_rest_default_user: cp-kafka-rest
kafka_rest_default_group: confluent
kafka_rest_default_log_dir: /var/log/confluent/kafka-rest
kafka_rest:
  server_start_file: "{{ binary_base_path }}/bin/kafka-rest-start"
  config_file: "{{ (config_base_path, kafka_rest_config_prefix_path, 'kafka-rest.properties') | path_join }}"
  systemd_file: "{{ (systemd_base_dir, (kafka_rest_service_name, '.service') | join) | path_join }}"
  systemd_override: "{{ ('/etc/systemd/system', (kafka_rest_service_name, '.service.d') | join , 'override.conf') | path_join }}"
  log4j_file: "{{ (base_path, 'etc/kafka-rest/log4j2.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, kafka_rest_config_prefix_path, 'jaas.conf') | path_join }}"
  password_file: "{{ (config_base_path, kafka_rest_config_prefix_path, 'password.properties') | path_join }}"

kafka_rest_http_protocol: "{{ 'https' if kafka_rest_ssl_enabled|bool else 'http' }}"

kafka_rest_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_rest_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_rest_kafka_listener_name]['port']}}"

kafka_rest_client_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[kafka_rest_client_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[kafka_rest_client_kafka_listener_name]['port']}}" # this needs to be added only in sasl_ssl kafka client i.e rp client

kafka_rest_properties:
  defaults:
    enabled: true
    properties:
      listeners: "{{kafka_rest_http_protocol}}://0.0.0.0:{{kafka_rest_port}}"
      host.name: "{{ hostvars[inventory_hostname]|confluent.platform.resolve_hostname }}"
      bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_rest_client_bootstrap_servers }}"
      confluent.license.topic: _confluent-command
  id:
    enabled: "{{ inventory_hostname in groups.kafka_rest }}"
    properties:
      # id logic depends on inventory hostname being in kafka_broker host list, defaulting to 0 if non rest proxy host
      id: "{{ groups.kafka_rest.index(inventory_hostname) + 1 if inventory_hostname in groups.kafka_rest else 0 }}"
  ssl:
    enabled: "{{kafka_rest_ssl_enabled}}"
    properties:
      ssl.keystore.location: "{{kafka_rest_keystore_path}}"
      ssl.keystore.password: "{{kafka_rest_keystore_storepass}}"
      ssl.key.password: "{{kafka_rest_keystore_keypass}}"
  truststore:
    enabled: "{{kafka_rest_ssl_mutual_auth_enabled|bool or (mds_tls_enabled|bool and not public_certificates_enabled|bool)}}"
    # CCloud kafka uses public certificates which are default java truststore
    properties:
      ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
  principal_mapping_rules:
    enabled: "{{kafka_rest_ssl_enabled}}"
    properties:
      auth.ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  mtls:
    enabled: "{{ kafka_rest_authentication_type == 'mtls' }}"
    properties:
      ssl.client.authentication: "{{ kafka_rest_ssl_client_authentication | upper }}"
  mds_mtls:
    # in case listener has oauth then we need to add oauth configs
    # if mtls only listener it will anyway not add oauth configs
    # this license client will use same configs as rest proxy in case there is no mtls on mds
    enabled: "{{ rbac_enabled|bool and mds_ssl_client_authentication != 'none' }}"
    properties: "{{ kafka_broker_listeners[kafka_rest_license_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.license.', kafka_rest_truststore_path, kafka_rest_truststore_storepass, public_certificates_enabled, kafka_rest_keystore_path, kafka_rest_keystore_storepass, kafka_rest_keystore_keypass,
                    false, sasl_plain_users_final.kafka_rest.principal, sasl_plain_users_final.kafka_rest.password, sasl_scram_users_final.kafka_rest.principal, sasl_scram_users_final.kafka_rest.password, sasl_scram256_users_final.kafka_rest.principal, sasl_scram256_users_final.kafka_rest.password,
                    kerberos_kafka_broker_primary, kafka_rest_keytab_path, kafka_rest_kerberos_principal|default('rp'),
                    false, kafka_rest_ldap_user, kafka_rest_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_rest_oauth_user, kafka_rest_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_rest_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_rest_third_party_oauth_client_assertion_config.license)) }}"
  kafka_license_bootstrap_server:
    enabled: "{{ rbac_enabled|bool and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.license.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_rest_bootstrap_servers }}"
  basic:
    enabled: "{{ kafka_rest_authentication_type == 'basic' }}"
    properties:
      authentication.method: BASIC
      authentication.realm: KafkaRest
      authentication.roles: "{{ kafka_rest_basic_users | confluent.platform.get_roles | unique | join(',') }}"
  kafka_client:
    enabled: true
    # ignore login callback handler related props using omit_oauth_configs when talking to internal_token listener
    properties: "{{ kafka_broker_listeners[kafka_rest_client_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                    'client.', kafka_rest_truststore_path, kafka_rest_truststore_storepass, public_certificates_enabled, kafka_rest_keystore_path, kafka_rest_keystore_storepass, kafka_rest_keystore_keypass,
                    false, sasl_plain_users_final.kafka_rest.principal, sasl_plain_users_final.kafka_rest.password, sasl_scram_users_final.kafka_rest.principal, sasl_scram_users_final.kafka_rest.password, sasl_scram256_users_final.kafka_rest.principal, sasl_scram256_users_final.kafka_rest.password,
                    kerberos_kafka_broker_primary, kafka_rest_keytab_path, kafka_rest_kerberos_principal|default('rp'),
                    kafka_rest_client_kafka_listener_name == 'internal_token' and auth_mode == 'mtls', kafka_rest_ldap_user, kafka_rest_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_rest_oauth_user, kafka_rest_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_rest_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_rest_third_party_oauth_client_assertion_config.kafka)) }}"
  kafka_client_password_protection:
    # Edge case for RP only- 'client' prefix not honored by secrets protection
    enabled: "{{ kafka_rest_secrets_protection_enabled }}"
    properties:
      client.config.providers: securepass
      client.config.providers.securepass.class: io.confluent.kafka.security.config.provider.SecurePassConfigProvider
  sr:
    enabled: "{{ 'schema_registry' in groups }}"
    properties:
      schema.registry.url: "{{schema_registry_url}}"
  sr_ssl:
    enabled: "{{ 'schema_registry' in groups and schema_registry_ssl_enabled|bool and not ccloud_schema_registry_enabled|bool }}"
    properties:
      schema.registry.ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      schema.registry.ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
      schema.registry.ssl.keystore.location: "{{kafka_rest_keystore_path}}"
      schema.registry.ssl.keystore.password: "{{kafka_rest_keystore_storepass}}"
      schema.registry.ssl.key.password: "{{kafka_rest_keystore_keypass}}"
  sr_basic:
    enabled: "{{ ('schema_registry' in groups and schema_registry_authentication_type == 'basic') or ccloud_schema_registry_enabled|bool }}"
    properties:
      basic.auth.credentials.source: USER_INFO
      schema.registry.basic.auth.user.info: "{{schema_registry_basic_users_final.admin.principal}}:{{schema_registry_basic_users_final.admin.password}}"
  monitoring_interceptor:
    enabled: "{{ kafka_rest_monitoring_interceptors_enabled|bool }}"
    properties:
      confluent.monitoring.interceptor.topic: _confluent-monitoring
      client.confluent.monitoring.interceptor.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else kafka_rest_bootstrap_servers }}"
  monitoring_interceptor_client:
    enabled: "{{ kafka_rest_monitoring_interceptors_enabled|bool }}"
    # ignore login callback handler related props using omit_oauth_configs when talking to internal_token listener
    properties: "{{ kafka_broker_listeners[kafka_rest_monitoring_interceptor_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                            'client.confluent.monitoring.interceptor.', kafka_rest_truststore_path, kafka_rest_truststore_storepass, public_certificates_enabled, kafka_rest_keystore_path, kafka_rest_keystore_storepass, kafka_rest_keystore_keypass,
                            false, sasl_plain_users_final.kafka_rest.principal, sasl_plain_users_final.kafka_rest.password, sasl_scram_users_final.kafka_rest.principal, sasl_scram_users_final.kafka_rest.password, sasl_scram256_users_final.kafka_rest.principal, sasl_scram256_users_final.kafka_rest.password,
                            kerberos_kafka_broker_primary, kafka_rest_keytab_path, kafka_rest_kerberos_principal|default('rp'),
                            kafka_rest_monitoring_interceptor_kafka_listener_name == 'internal_token' and auth_mode == 'mtls', kafka_rest_ldap_user, kafka_rest_ldap_password, mds_bootstrap_server_urls, oauth_enabled, kafka_rest_oauth_user, kafka_rest_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_rest_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_rest_third_party_oauth_client_assertion_config.monitoring_interceptor)) }}"
  rbac:
    enabled: "{{rbac_enabled}}"
    properties:
      kafka.rest.resource.extension.class: io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      public.key.path: "{{rbac_enabled_public_pem_path}}"
      confluent.metadata.bootstrap.server.urls: "{{mds_bootstrap_server_urls}}"
  rbac_ldap:
    enabled: "{{ rbac_enabled and auth_mode == 'ldap' and (not kafka_rest_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.http.auth.credentials.provider: BASIC
      confluent.metadata.basic.auth.user.info: "{{kafka_rest_ldap_user | default('rest') }}:{{kafka_rest_ldap_password | default('pass')}}"
  rbac_oauth:
    enabled: "{{ rbac_enabled and oauth_enabled and (not kafka_rest_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.http.auth.credentials.provider: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.id: "{{kafka_rest_oauth_user}}"
      confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
  rbac_oauth_secret:
    enabled: "{{ rbac_enabled and oauth_enabled and (not kafka_rest_mds_cert_auth_only) and (not kafka_rest_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.secret: "{{kafka_rest_oauth_password}}"
  rbac_oauth_scope:
    enabled: "{{rbac_enabled and oauth_enabled and oauth_groups_scope!='none' and (not kafka_rest_mds_cert_auth_only) and (not kafka_rest_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.oauth.scope: "{{oauth_groups_scope}}"
  rbac_oauth_third_party_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_rest_oauth_client_assertion_enabled and kafka_rest_oauth_client_assertion_file_base_path !='none' and (not kafka_rest_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.assertion.file: "{{ kafka_rest_third_party_oauth_client_assertion_config.mds }}"
  rbac_oauth_local_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_rest_oauth_client_assertion_enabled and kafka_rest_oauth_client_assertion_file_base_path =='none' and (not kafka_rest_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.metadata.oauthbearer.assertion.claim.iss: "{{kafka_rest_oauth_client_assertion_issuer}}"
      confluent.metadata.oauthbearer.assertion.claim.sub: "{{kafka_rest_oauth_client_assertion_sub}}"
      confluent.metadata.oauthbearer.assertion.claim.aud: "{{kafka_rest_oauth_client_assertion_audience}}"
      confluent.metadata.oauthbearer.assertion.private.key.file: "{{ kafka_rest_oauth_client_assertion_private_key_file_dest_path }}"
  rbac_oauth_local_client_assertion_template_file:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_rest_oauth_client_assertion_enabled and kafka_rest_oauth_client_assertion_file_base_path =='none' and (not kafka_rest_mds_cert_auth_only) and kafka_rest_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.template.file: "{{ kafka_rest_oauth_client_assertion_template_file_dest_path }}"
  rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_rest_oauth_client_assertion_enabled and kafka_rest_oauth_client_assertion_file_base_path =='none' and (not kafka_rest_mds_cert_auth_only) and kafka_rest_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{kafka_rest_oauth_client_assertion_private_key_passphrase}}"
  rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_rest_oauth_client_assertion_enabled and kafka_rest_oauth_client_assertion_file_base_path =='none' and (not kafka_rest_mds_cert_auth_only) and kafka_rest_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{kafka_rest_oauth_client_assertion_jti_include}}"
  rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{rbac_enabled and oauth_enabled and kafka_rest_oauth_client_assertion_enabled and kafka_rest_oauth_client_assertion_file_base_path =='none' and (not kafka_rest_mds_cert_auth_only) and kafka_rest_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{kafka_rest_oauth_client_assertion_nbf_include}}"
  kafka_rest_oauth:
    enabled: "{{kafka_rest_oauth_enabled}}"
    properties:
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      oauthbearer.jwks.endpoint.url: "{{oauth_jwks_uri}}"
      oauthbearer.sub.claim.name: "{{ oauth_sub_claim }}"
      oauthbearer.expected.issuer: "{{ oauth_issuer_url }}"
  kafka_rest_oauth_claim:
    enabled: "{{kafka_rest_oauth_enabled and oauth_groups_claim!='none' }}"
    properties:
      oauthbearer.groups.claim.name: "{{ oauth_groups_claim }}"
  kafka_rest_oauth_audience:
    enabled: "{{kafka_rest_oauth_enabled and oauth_expected_audience!='none' }}"
    properties:
      oauthbearer.expected.audience: "{{ oauth_expected_audience }}"
  kafka_rest_oauth_ssl:
    enabled: "{{kafka_rest_oauth_enabled and idp_self_signed}}"
    properties:
      ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
  rbac_external_client:
    enabled: "{{rbac_enabled and external_mds_enabled and mds_tls_enabled }}"
    properties:
      ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
      client.ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      client.ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
  mds_client:
    enabled: "{{rbac_enabled and mds_tls_enabled }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
      confluent.metadata.ssl.endpoint.identification.algorithm: HTTPS
  mds_mtls_client:
    enabled: "{{ rbac_enabled and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.ssl.keystore.location: "{{kafka_rest_keystore_path}}"
      confluent.metadata.ssl.keystore.password: "{{kafka_rest_keystore_keypass}}"
      confluent.metadata.ssl.key.password: "{{kafka_rest_keystore_keypass}}"
      confluent.metadata.ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
  oauth_mds_client:
    enabled: "{{oauth_enabled and rbac_enabled and idp_self_signed }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{kafka_rest_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_rest_truststore_storepass}}"
  telemetry:
    enabled: "{{kafka_rest_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter
  telemetry_proxy:
    enabled: "{{kafka_rest_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{kafka_rest_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{kafka_rest_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"

kafka_rest_combined_properties: "{{kafka_rest_properties | confluent.platform.combine_properties}}"

kafka_rest_final_properties: "{{ kafka_rest_combined_properties | combine(kafka_rest_custom_properties) }}"

### Control Center Next Generation Variables ###
control_center_next_gen_service_name: confluent-control-center
control_center_next_gen_dep_prometheus_service_name: prometheus
control_center_next_gen_dep_alertmanager_service_name: alertmanager
control_center_next_gen_default_user: cp-control-center
control_center_next_gen_default_group: confluent
control_center_next_gen_default_log_dir: /var/log/confluent/control-center
control_center_next_gen_dep_prometheus:
  config_path: "{{control_center_next_gen_dependencies_config_path}}"
  data_path: "{{control_center_next_gen_data_dir}}/prometheus"
  log_path: "{{control_center_next_gen_log_dir}}"
  server_start_file: "{{ confluent_control_center_next_gen_binary_base_path }}/bin/prometheus-start"
  config_file: "{{control_center_next_gen_dependencies_config_path}}/prometheus-generated.yml"
  recording_rules_config_file: "{{control_center_next_gen_dependencies_config_path}}/recording_rules-generated.yml"
  trigger_rules_config_file: "{{control_center_next_gen_dependencies_config_path}}/trigger_rules-generated.yml"
  web_config_file: "{{control_center_next_gen_dependencies_config_path}}/web-config-prom.yml"
  systemd_file: "{{ (systemd_base_dir, (control_center_next_gen_dep_prometheus_service_name, '.service') | join) | path_join }}"
  systemd_override: "{{ ('/etc/systemd/system', (control_center_next_gen_dep_prometheus_service_name, '.service.d') | join , 'override.conf') | path_join }}"
  prometheus_web_listener_host: "{{ control_center_next_gen_dependency_prometheus_web_listener_host }}"
  prometheus_port: "{{ control_center_next_gen_dependency_prometheus_port }}"
control_center_next_gen_dep_alertmanager:
  config_path: "{{control_center_next_gen_dependencies_config_path}}"
  data_path: "{{control_center_next_gen_data_dir}}/alertmanager"
  log_path: "{{control_center_next_gen_log_dir}}"
  server_start_file: "{{ confluent_control_center_next_gen_binary_base_path }}/bin/alertmanager-start"
  config_file: "{{control_center_next_gen_dependencies_config_path}}/alertmanager-generated.yml"
  web_config_file: "{{control_center_next_gen_dependencies_config_path}}/web-config-am.yml"
  systemd_file: "{{ (systemd_base_dir, (control_center_next_gen_dep_alertmanager_service_name, '.service') | join) | path_join }}"
  systemd_override: "{{ ('/etc/systemd/system', (control_center_next_gen_dep_alertmanager_service_name, '.service.d') | join , 'override.conf') | path_join }}"
  alertmanager_web_listener_host: "{{ control_center_next_gen_dependency_alertmanager_web_listener_host }}"
  alertmanager_port: "{{ control_center_next_gen_dependency_alertmanager_port }}"
control_center_next_gen:
  server_start_file: "{{ confluent_control_center_next_gen_binary_base_path }}/bin/control-center-start"
  config_file: "{{ (config_base_path, control_center_next_gen_config_prefix_path, 'control-center-production.properties') | path_join }}"
  systemd_file: "{{ (systemd_base_dir, (control_center_next_gen_service_name, '.service') | join) | path_join }}"
  systemd_override: "{{ ('/etc/systemd/system', (control_center_next_gen_service_name, '.service.d') | join , 'override.conf') | path_join }}"
  log4j_file: "{{ (confluent_control_center_next_gen_base_path, 'etc/confluent-control-center/log4j2-rolling.yaml') | path_join }}"
  jaas_file: "{{ (config_base_path, control_center_next_gen_config_prefix_path, 'jaas.conf') | path_join }}"
  password_file: "{{ (config_base_path, control_center_next_gen_config_prefix_path, 'password.properties') | path_join }}"

control_center_next_gen_http_protocol: "{{ 'https' if control_center_next_gen_ssl_enabled|bool else 'http' }}"

control_center_next_gen_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[control_center_next_gen_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[control_center_next_gen_kafka_listener_name]['port']}}"

control_center_next_gen_streams_bootstrap_servers: "{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_hostnames(hostvars) | join(':' + kafka_broker_listeners[control_center_next_gen_streams_kafka_listener_name]['port']|string + ',') }}:{{kafka_broker_listeners[control_center_next_gen_streams_kafka_listener_name]['port']}}" # this needs to be added only in sasl_ssl kafka client i.e streams

control_center_next_gen_properties:
  defaults:
    enabled: true
    properties:
      confluent.controlcenter.streams.num.stream.threads: 12
      confluent.controlcenter.data.dir: "{{control_center_next_gen_data_dir}}"
      confluent.controlcenter.internal.topics.replication: "{{control_center_next_gen_default_internal_replication_factor}}"
      confluent.metrics.topic.replication: "{{control_center_next_gen_default_internal_replication_factor}}"
      confluent.monitoring.interceptor.topic.replication: "{{control_center_next_gen_default_internal_replication_factor}}"
      confluent.controlcenter.command.topic.replication: "{{control_center_next_gen_default_internal_replication_factor}}"
      confluent.controlcenter.rest.listeners: "{{control_center_next_gen_http_protocol}}://{{control_center_next_gen_listener_hostname | confluent.platform.format_hostname}}:{{control_center_next_gen_port}}"
      bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else control_center_next_gen_streams_bootstrap_servers }}"
      confluent.controlcenter.streams.security.protocol: "{{kafka_broker_listeners[control_center_next_gen_streams_kafka_listener_name] | confluent.platform.kafka_protocol_defaults(ssl_enabled, sasl_protocol) }}"
      confluent.controlcenter.prometheus.enable: true
      confluent.controlcenter.prometheus.url: "{{control_center_next_gen_dependency_prometheus_url}}"
      confluent.controlcenter.alertmanager.url: "{{control_center_next_gen_dependency_alertmanager_url}}"
      confluent.controlcenter.prometheus.config.file: "{{control_center_next_gen_dep_prometheus.config_file}}"
      confluent.controlcenter.prometheus.rules.file: "{{control_center_next_gen_dep_prometheus.trigger_rules_config_file}}"
      confluent.controlcenter.alertmanager.config.file: "{{control_center_next_gen_dep_alertmanager.config_file}}"
  prometheus_client:
    enabled: "{{control_center_next_gen_dependency_prometheus_ssl_enabled | bool or control_center_next_gen_dependency_prometheus_basic_auth_enabled | bool}}"
    properties: "{{ control_center_next_gen_dependency_prometheus_ssl_enabled | confluent.platform.prometheus_client_properties(
        control_center_next_gen_dependency_prometheus_truststore_path,
        control_center_next_gen_dependency_prometheus_truststore_storepass,
        control_center_next_gen_dependency_prometheus_keystore_path,
        control_center_next_gen_dependency_prometheus_keystore_storepass,
        control_center_next_gen_dependency_prometheus_keystore_keypass,
        control_center_next_gen_dependency_prometheus_basic_auth_enabled,
        kafka_broker_telemetry_control_center_next_gen_user + ':' + kafka_broker_telemetry_control_center_next_gen_password,
        control_center_next_gen_dependency_prometheus_mtls_enabled,
        control_center_next_gen_dependency_prometheus_service_name
      ) }}"
  alertmanager_client:
    enabled: "{{control_center_next_gen_dependency_alertmanager_ssl_enabled | bool or control_center_next_gen_dependency_alertmanager_basic_auth_enabled | bool}}"
    properties: "{{ control_center_next_gen_dependency_alertmanager_ssl_enabled | confluent.platform.alertmanager_client_properties(
        control_center_next_gen_dependency_alertmanager_truststore_path,
        control_center_next_gen_dependency_alertmanager_truststore_storepass,
        control_center_next_gen_dependency_alertmanager_keystore_path,
        control_center_next_gen_dependency_alertmanager_keystore_storepass,
        control_center_next_gen_dependency_alertmanager_keystore_keypass,
        control_center_next_gen_dependency_alertmanager_basic_auth_enabled,
        control_center_next_gen_dependency_alertmanager_basic_users.admin.principal + ':' + control_center_next_gen_dependency_alertmanager_basic_users.admin.password,
        control_center_next_gen_dependency_alertmanager_mtls_enabled,
        control_center_next_gen_dependency_alertmanager_service_name
      ) }}"
  ssl:
    enabled: "{{control_center_next_gen_ssl_enabled or (kafka_broker_rest_proxy_enabled and mds_tls_enabled) }}"
    properties:
      confluent.controlcenter.rest.ssl.keystore.location: "{{control_center_next_gen_keystore_path}}"
      confluent.controlcenter.rest.ssl.keystore.password: "{{control_center_next_gen_keystore_storepass}}"
      confluent.controlcenter.rest.ssl.key.password: "{{control_center_next_gen_keystore_keypass}}"
      confluent.controlcenter.rest.ssl.truststore.location: "{{control_center_next_gen_truststore_path}}"
      confluent.controlcenter.rest.ssl.truststore.password: "{{control_center_next_gen_truststore_storepass}}"
  principal_mapping_rules:
    enabled: "{{control_center_next_gen_ssl_enabled}}"
    properties:
      ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  basic:
    enabled: "{{ control_center_next_gen_authentication_type == 'basic' }}"
    properties:
      confluent.controlcenter.rest.authentication.method: BASIC
      confluent.controlcenter.rest.authentication.realm: ControlCenter
      confluent.controlcenter.rest.authentication.roles: "{{ (control_center_next_gen_basic_users|confluent.platform.get_roles + [ 'Restricted' ]) | unique | join(',') }}"
      confluent.controlcenter.auth.restricted.roles: Restricted
      confluent.controlcenter.auth.session.expiration.ms: 600000
  broker_embedded_rest_endpoint:
    enabled: "{{kafka_broker_rest_proxy_enabled or rbac_enabled }}"
    properties:
      confluent.controlcenter.streams.cprest.url: "{{mds_http_protocol}}://{{ groups['kafka_broker'] | default(['localhost']) | confluent.platform.resolve_and_format_hostnames(hostvars) | join(':' + mds_port|string + ',' + mds_http_protocol + '://') }}:{{mds_port}}"
  kafka_client:
    enabled: true # ignore login callback handler related props using omit_oauth_configs when talking to internal_token listener
    properties: "{{ kafka_broker_listeners[control_center_next_gen_streams_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.controlcenter.streams.', control_center_next_gen_truststore_path, control_center_next_gen_truststore_storepass, public_certificates_enabled, control_center_next_gen_keystore_path, control_center_next_gen_keystore_storepass, control_center_next_gen_keystore_keypass,
                    false, sasl_plain_users_final.control_center_next_gen.principal, sasl_plain_users_final.control_center_next_gen.password, sasl_scram_users_final.control_center_next_gen.principal, sasl_scram_users_final.control_center_next_gen.password, sasl_scram256_users_final.control_center_next_gen.principal, sasl_scram256_users_final.control_center_next_gen.password,
                    kerberos_kafka_broker_primary, control_center_next_gen_keytab_path, control_center_next_gen_kerberos_principal|default('c3'),
                    control_center_next_gen_streams_kafka_listener_name == 'internal_token', control_center_next_gen_ldap_user, control_center_next_gen_ldap_password, mds_bootstrap_server_urls, oauth_enabled, control_center_next_gen_oauth_user, control_center_next_gen_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, control_center_next_gen_oauth_client_assertion_config) }}"
  kafka_client_token_cert_lcbh:  # add the new login callback handler for c3 to kafka communication
    enabled: "{{ control_center_next_gen_streams_kafka_listener_name == 'internal_token' }}"
    properties:
      confluent.controlcenter.streams.sasl.login.callback.handler.class: io.confluent.kafka.clients.plugins.auth.token.TokenCertificateLoginCallbackHandler
  kafka_interceptors:
    enabled: true # ignore login callback handler related props using omit_oauth_configs when talking to internal_token listener
    properties: "{{ kafka_broker_listeners[control_center_next_gen_monitoring_interceptor_kafka_listener_name] | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                    'confluent.monitoring.interceptor.', control_center_next_gen_truststore_path, control_center_next_gen_truststore_storepass, public_certificates_enabled, control_center_next_gen_keystore_path, control_center_next_gen_keystore_storepass, control_center_next_gen_keystore_keypass,
                    false, sasl_plain_users_final.control_center_next_gen.principal, sasl_plain_users_final.control_center_next_gen.password, sasl_scram_users_final.control_center_next_gen.principal, sasl_scram_users_final.control_center_next_gen.password, sasl_scram256_users_final.control_center_next_gen.principal, sasl_scram256_users_final.control_center_next_gen.password,
                    kerberos_kafka_broker_primary, control_center_next_gen_keytab_path, control_center_next_gen_kerberos_principal|default('c3'),
                    control_center_next_gen_monitoring_interceptor_kafka_listener_name == 'internal_token', control_center_next_gen_ldap_user, control_center_next_gen_ldap_password, mds_bootstrap_server_urls, oauth_enabled, control_center_next_gen_oauth_user, control_center_next_gen_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, control_center_next_gen_oauth_client_assertion_config) }}"
  kafka_interceptors_bootstrap_server:
    enabled: true
    properties:
      confluent.monitoring.interceptor.bootstrap.servers: "{{ ccloud_kafka_bootstrap_servers if ccloud_kafka_enabled|bool else control_center_next_gen_bootstrap_servers }}"
  kafka_interceptors_token_cert_lcbh: # add the new login callback handler for c3 to kafka communication
    enabled: "{{ control_center_next_gen_monitoring_interceptor_kafka_listener_name == 'internal_token' }}"
    properties:
      confluent.monitoring.interceptor.sasl.login.callback.handler.class: io.confluent.kafka.clients.plugins.auth.token.TokenCertificateLoginCallbackHandler
  sr:
    enabled: "{{ 'schema_registry' in groups }}"
    properties:
      confluent.controlcenter.schema.registry.url: "{{schema_registry_url}}"
  sr_ssl:
    enabled: "{{ 'schema_registry' in groups and schema_registry_ssl_enabled|bool and not ccloud_schema_registry_enabled|bool }}"
    properties:
      confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.location: "{{control_center_next_gen_truststore_path}}"
      confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.password: "{{control_center_next_gen_truststore_storepass}}"
      confluent.controlcenter.schema.registry.schema.registry.ssl.keystore.location: "{{control_center_next_gen_keystore_path}}"
      confluent.controlcenter.schema.registry.schema.registry.ssl.keystore.password: "{{control_center_next_gen_keystore_storepass}}"
      confluent.controlcenter.schema.registry.schema.registry.ssl.key.password: "{{control_center_next_gen_keystore_keypass}}"
  sr_basic:
    enabled: "{{ ('schema_registry' in groups and schema_registry_authentication_type == 'basic') or ccloud_schema_registry_enabled|bool }}"
    properties:
      confluent.controlcenter.schema.registry.basic.auth.credentials.source: USER_INFO
      confluent.controlcenter.schema.registry.basic.auth.user.info: "{{schema_registry_basic_users_final.admin.principal}}:{{schema_registry_basic_users_final.admin.password}}"
  sr_oauth:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.login.client.id: "{{ control_center_next_gen_oauth_user }}"
  sr_oauth_secret:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and (not control_center_next_gen_oauth_client_assertion_enabled) }}"
    properties:
      confluent.controlcenter.schema.registry.bearer.auth.credentials.source: OAUTHBEARER
      confluent.controlcenter.schema.registry.oauthbearer.login.client.secret: "{{ control_center_next_gen_oauth_password }}"
  sr_oauth_scope:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and oauth_groups_scope!='none' and (not control_center_next_gen_oauth_client_assertion_enabled) }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.login.oauth.scope: "{{ oauth_groups_scope }}"
  sr_oauth_third_party_client_assertion:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path != 'none' }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.assertion.file: "{{ control_center_next_gen_third_party_oauth_client_assertion_config.kafka }}"
  sr_oauth_local_client_assertion:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' }}"
    properties:
      confluent.controlcenter.schema.registry.bearer.auth.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.controlcenter.schema.registry.oauthbearer.assertion.claim.iss: "{{ control_center_next_gen_oauth_client_assertion_issuer }}"
      confluent.controlcenter.schema.registry.oauthbearer.assertion.claim.sub: "{{ control_center_next_gen_oauth_client_assertion_sub }}"
      confluent.controlcenter.schema.registry.oauthbearer.assertion.claim.aud: "{{ control_center_next_gen_oauth_client_assertion_audience }}"
      confluent.controlcenter.schema.registry.oauthbearer.assertion.private.key.file: "{{ control_center_next_gen_oauth_client_assertion_private_key_file_dest_path }}"
  sr_oauth_local_client_assertion_template_file:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and control_center_next_gen_oauth_client_assertion_template_file != 'none' }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.assertion.template.file: "{{ control_center_next_gen_oauth_client_assertion_template_file_dest_path }}"
  sr_oauth_local_client_assertion_passphrase:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and control_center_next_gen_oauth_client_assertion_private_key_passphrase != 'none' }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.assertion.private.key.passphrase: "{{ control_center_next_gen_oauth_client_assertion_private_key_passphrase }}"
  sr_oauth_local_client_assertion_jti_include:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and control_center_next_gen_oauth_client_assertion_jti_include != 'none' }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.assertion.claim.jti.include: "{{ control_center_next_gen_oauth_client_assertion_jti_include }}"
  sr_oauth_local_client_assertion_nbf_include:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and control_center_next_gen_oauth_client_assertion_nbf_include != 'none' }}"
    properties:
      confluent.controlcenter.schema.registry.oauthbearer.assertion.claim.nbf.include: "{{ control_center_next_gen_oauth_client_assertion_nbf_include }}"
  sr_oauth_ssl:
    enabled: "{{ ('schema_registry' in groups and schema_registry_oauth_enabled|bool) and (not rbac_enabled|bool) and idp_self_signed }}"
    properties:
      confluent.controlcenter.schema.registry.ssl.truststore.location: "{{ control_center_next_gen_truststore_path }}"
      confluent.controlcenter.schema.registry.ssl.truststore.password: "{{ control_center_next_gen_truststore_storepass }}"
  connect:
    enabled: "{{ kafka_connect_cluster_ansible_group_names | default(omit) | length > 0 }}"
    properties: "{{ kafka_connect_cluster_ansible_group_names | confluent.platform.c3_connect_properties(groups, hostvars, kafka_connect_ssl_enabled,
        kafka_connect_http_protocol, kafka_connect_rest_port, kafka_connect_group_id, control_center_next_gen_truststore_path, control_center_next_gen_truststore_storepass,
        control_center_next_gen_keystore_path, control_center_next_gen_keystore_storepass, control_center_next_gen_keystore_keypass, kafka_connect_oauth_enabled, rbac_enabled,
        control_center_next_gen_oauth_user, control_center_next_gen_oauth_password, oauth_groups_scope, idp_self_signed) }}"
  ksql:
    enabled: "{{ ksql_cluster_ansible_group_names | default(omit) | length > 0 }}"
    properties: "{{ ksql_cluster_ansible_group_names | confluent.platform.c3_ksql_properties(groups, hostvars, ksql_ssl_enabled,
        ksql_http_protocol, ksql_listener_port, control_center_next_gen_truststore_path, control_center_next_gen_truststore_storepass,
        control_center_next_gen_keystore_path, control_center_next_gen_keystore_storepass, control_center_next_gen_keystore_keypass, ksql_oauth_enabled, rbac_enabled,
        control_center_next_gen_oauth_user, control_center_next_gen_oauth_password, oauth_groups_scope, idp_self_signed) }}"
  oauth:
    enabled: "{{oauth_enabled or schema_registry_oauth_enabled or kafka_connect_oauth_enabled or ksql_oauth_enabled}}"
    properties:
      confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
  rbac:
    enabled: "{{rbac_enabled}}"
    properties:
      confluent.controlcenter.rest.authentication.method: BEARER
      public.key.path: "{{rbac_enabled_public_pem_path}}"
      confluent.metadata.bootstrap.server.urls: "{{mds_bootstrap_server_urls}}"
  rbac_ldap:
    enabled: "{{rbac_enabled and auth_mode == 'ldap' and (not control_center_next_gen_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.basic.auth.user.info: "{{ control_center_next_gen_ldap_user | default('c3') }}:{{ control_center_next_gen_ldap_password | default('pass') }}"
      confluent.metadata.http.auth.credentials.provider: BASIC
  rbac_oauth:
    enabled: "{{rbac_enabled and oauth_enabled and (not control_center_next_gen_mds_cert_auth_only)}}"
    properties:
      confluent.metadata.oauthbearer.login.client.id: "{{ control_center_next_gen_oauth_user }}"
  rbac_oauth_secret:
    enabled: "{{rbac_enabled and oauth_enabled and (not control_center_next_gen_mds_cert_auth_only) and (not control_center_next_gen_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.client.secret: "{{ control_center_next_gen_oauth_password }}"
  rbac_oauth_scope:
    enabled: "{{rbac_enabled and oauth_enabled and oauth_groups_scope != 'none' and (not control_center_next_gen_mds_cert_auth_only) and (not control_center_next_gen_oauth_client_assertion_enabled)}}"
    properties:
      confluent.metadata.oauthbearer.login.oauth.scope: "{{ oauth_groups_scope }}"
  rbac_oauth_third_party_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path != 'none' and (not control_center_next_gen_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.assertion.file: "{{ control_center_next_gen_third_party_oauth_client_assertion_config.kafka }}"
  rbac_oauth_local_client_assertion:
    enabled: "{{rbac_enabled and oauth_enabled and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and (not control_center_next_gen_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.iss: "{{ control_center_next_gen_oauth_client_assertion_issuer }}"
      confluent.metadata.oauthbearer.assertion.claim.sub: "{{ control_center_next_gen_oauth_client_assertion_sub }}"
      confluent.metadata.oauthbearer.assertion.claim.aud: "{{ control_center_next_gen_oauth_client_assertion_audience }}"
      confluent.metadata.oauthbearer.assertion.private.key.file: "{{ control_center_next_gen_oauth_client_assertion_private_key_file_dest_path }}"
  rbac_oauth_local_client_assertion_template_file:
    enabled: "{{rbac_enabled and oauth_enabled and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and (not control_center_next_gen_mds_cert_auth_only) and control_center_next_gen_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.template.file: "{{ control_center_next_gen_oauth_client_assertion_template_file_dest_path }}"
  rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{rbac_enabled and oauth_enabled and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and (not control_center_next_gen_mds_cert_auth_only) and control_center_next_gen_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{ control_center_next_gen_oauth_client_assertion_private_key_passphrase }}"
  rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{rbac_enabled and oauth_enabled and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and (not control_center_next_gen_mds_cert_auth_only) and control_center_next_gen_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{ control_center_next_gen_oauth_client_assertion_jti_include }}"
  rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{rbac_enabled and oauth_enabled and control_center_next_gen_oauth_client_assertion_enabled and control_center_next_gen_oauth_client_assertion_file_base_path == 'none' and (not control_center_next_gen_mds_cert_auth_only) and control_center_next_gen_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{ control_center_next_gen_oauth_client_assertion_nbf_include }}"
  rbac_external_client:
    enabled: "{{rbac_enabled and external_mds_enabled and mds_tls_enabled }}"
    properties:
      confluent.controlcenter.streams.ssl.truststore.location: "{{control_center_next_gen_truststore_path}}"
      confluent.controlcenter.streams.ssl.truststore.password: "{{control_center_next_gen_truststore_storepass}}"
  rbac_mds_sso:
    enabled: "{{ rbac_enabled and sso_mode != 'none' }}"
    properties:
      confluent.controlcenter.auth.sso.mode: "{{ sso_mode }}"
  mds_client:
    enabled: "{{rbac_enabled and mds_tls_enabled }}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{control_center_next_gen_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{control_center_next_gen_truststore_storepass}}"
      confluent.metadata.ssl.endpoint.identification.algorithm: HTTPS
  mds_mtls_client:
    enabled: "{{ rbac_enabled and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.ssl.keystore.location: "{{control_center_next_gen_keystore_path}}"
      confluent.metadata.ssl.keystore.password: "{{control_center_next_gen_keystore_keypass}}"
      confluent.metadata.ssl.key.password: "{{control_center_next_gen_keystore_keypass}}"
      confluent.metadata.ssl.truststore.location: "{{control_center_next_gen_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{control_center_next_gen_truststore_storepass}}"
  oauth_mds_client:
    enabled: "{{rbac_enabled and oauth_enabled and idp_self_signed}}"
    properties:
      confluent.metadata.ssl.truststore.location: "{{ control_center_next_gen_truststore_path }}"
      confluent.metadata.ssl.truststore.password: "{{ control_center_next_gen_truststore_storepass }}"
  telemetry:
    enabled: "{{control_center_next_gen_telemetry_enabled}}"
    properties:
      confluent.telemetry.enabled: 'true'
      confluent.telemetry.api.key: "{{telemetry_api_key}}"
      confluent.telemetry.api.secret: "{{telemetry_api_secret}}"
      metric.reporters: io.confluent.telemetry.reporter.TelemetryReporter
  telemetry_proxy:
    enabled: "{{control_center_next_gen_telemetry_enabled and telemetry_proxy_url != ''}}"
    properties:
      confluent.telemetry.proxy.url: "{{telemetry_proxy_url}}"
  telemetry_proxy_auth:
    enabled: "{{control_center_next_gen_telemetry_enabled and telemetry_proxy_username != ''}}"
    properties:
      confluent.telemetry.proxy.username: "{{telemetry_proxy_username}}"
      confluent.telemetry.proxy.password: "{{telemetry_proxy_password}}"
  telemetry_labels:
    enabled: "{{control_center_next_gen_telemetry_ansible_labels_enabled}}"
    properties:
      confluent.telemetry.labels.confluent.ansible.playbooks.version: "{{confluent_ansible_branch}}"
  ccloud:
    enabled: "{{ccloud_kafka_enabled}}"
    properties:
      confluent.metrics.topic.max.message.bytes: 8388608

control_center_next_gen_dependency_prometheus_basic_auth_users: "{{ control_center_next_gen_dependency_prometheus_basic_users | confluent.platform.c3_generate_salt_and_hash}}"

control_center_next_gen_dependency_alertmanager_basic_auth_users: "{{ control_center_next_gen_dependency_alertmanager_basic_users | confluent.platform.c3_generate_salt_and_hash}}"

control_center_next_gen_combined_properties: "{{control_center_next_gen_properties | confluent.platform.combine_properties}}"

control_center_next_gen_final_properties: "{{ control_center_next_gen_combined_properties | combine(control_center_next_gen_custom_properties) }}"

#### Kafka Connect Replicator Variables ####
kafka_connect_replicator_service_name: kafka-connect-replicator
kafka_connect_replicator_default_user: cp-kafka-connect-replicator
kafka_connect_replicator_default_group: confluent

kafka_connect_replicator:
  server_start_file: "{{ binary_base_path }}/bin/kafka-connect-replicator-start"
  replication_config_file: "{{ (config_base_path, kafka_connect_replicator_config_prefix_path, 'kafka-connect-replicator.properties') | path_join }}"
  consumer_config_file: "{{ (config_base_path, kafka_connect_replicator_config_prefix_path, 'kafka-connect-replicator-consumer.properties') | path_join }}"
  producer_config_file: "{{ (config_base_path, kafka_connect_replicator_config_prefix_path, 'kafka-connect-replicator-producer.properties') | path_join }}"
  interceptors_config_file: "{{ (config_base_path, kafka_connect_replicator_config_prefix_path, 'kafka-connect-replicator-interceptors.properties') | path_join }}"
  systemd_file: "{{ (systemd_base_dir, (kafka_connect_replicator_service_name, '.service') | join) | path_join }}"
  systemd_override: "{{ ('/etc/systemd/system', (kafka_connect_replicator_service_name, '.service.d') | join , 'override.conf') | path_join }}"
  log4j_file: "{{ (base_path, 'etc/kafka-connect-replicator/replicator-log4j2.yaml') | path_join }}"

kafka_connect_replicator_rest_extension_classes:
  - "{% if kafka_connect_replicator_rbac_enabled|bool %}io.confluent.connect.security.ConnectSecurityExtension{% endif %}"

kafka_connect_replicator_final_rest_extension_classes: "{{(kafka_connect_replicator_rest_extension_classes|difference(['']) + kafka_connect_replicator_custom_rest_extension_classes) | unique}}"

kafka_connect_replicator_properties:
  defaults:
    enabled: true
    properties:
      listeners: "{{kafka_connect_replicator_http_protocol}}://0.0.0.0:{{kafka_connect_replicator_port}}"
      rest.advertised.listener: "{{kafka_connect_replicator_http_protocol}}"
      rest.advertised.host.name: "{{ hostvars[inventory_hostname]|confluent.platform.resolve_hostname }}"
      rest.advertised.port: "{{kafka_connect_replicator_port}}"
      group.id: "{{kafka_connect_replicator_group_id}}"
      confluent.topic.bootstrap.servers: "{{kafka_connect_replicator_bootstrap_servers}}"
      offset.start: "{{kafka_connect_replicator_offset_start}}"
      offset.storage.topic: "{{kafka_connect_replicator_offsets_topic}}"
      status.storage.topic: "{{kafka_connect_replicator_status_topic}}"
      config.storage.topic: "{{kafka_connect_replicator_storage_topic}}"
      topic.whitelist: "{{kafka_connect_replicator_white_list}}"
      topic.auto.create: "{{kafka_connect_replicator_topic_auto_create}}"
      confluent.license.topic: _confluent-command
  rest_classes:
    enabled: "{{ kafka_connect_replicator_final_rest_extension_classes|length > 0 }}"
    properties:
      rest.extension.classes: "{{ kafka_connect_replicator_final_rest_extension_classes | join(',') }}"
  ssl:
    enabled: "{{kafka_connect_replicator_ssl_enabled}}"
    properties:
      listeners.https.ssl.keystore.location: "{{kafka_connect_replicator_keystore_path}}"
      listeners.https.ssl.keystore.password: "{{kafka_connect_replicator_keystore_storepass}}"
      listeners.https.ssl.key.password: "{{kafka_connect_replicator_keystore_keypass}}"
      listeners.https.ssl.truststore.location: "{{kafka_connect_replicator_truststore_path}}"
      listeners.https.ssl.truststore.password: "{{kafka_connect_replicator_truststore_storepass}}"
  principal_mapping_rules:
    enabled: "{{ssl_enabled}}"
    properties:
      auth.ssl.principal.mapping.rules: "{{ principal_mapping_rules | join(',') }}"
  mds_mtls_client:
    enabled: "{{ kafka_connect_replicator_rbac_enabled and mds_ssl_client_authentication != 'none' }}"
    properties:
      confluent.metadata.ssl.keystore.location: "{{kafka_connect_replicator_keystore_path}}"
      confluent.metadata.ssl.keystore.password: "{{kafka_connect_replicator_keystore_keypass}}"
      confluent.metadata.ssl.key.password: "{{kafka_connect_replicator_keystore_keypass}}"
      confluent.metadata.ssl.truststore.location: "{{kafka_connect_replicator_truststore_path}}"
      confluent.metadata.ssl.truststore.password: "{{kafka_connect_replicator_truststore_storepass}}"
  rbac:
    enabled: "{{kafka_connect_replicator_rbac_enabled}}"
    properties:
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      public.key.path: "{{kafka_connect_replicator_rbac_enabled_public_pem_path}}"
      confluent.metadata.bootstrap.server.urls: "{{kafka_connect_replicator_erp_host}}"
  rbac_ldap:
    enabled: "{{ kafka_connect_replicator_rbac_enabled and auth_mode == 'ldap'  and (not kafka_connect_replicator_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.basic.auth.user.info: "{{kafka_connect_replicator_ldap_user | default('replicator') }}:{{kafka_connect_replicator_ldap_password | default('pass')}}"
      confluent.metadata.http.auth.credentials.provider: BASIC
  rbac_oauth:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and (not kafka_connect_replicator_mds_cert_auth_only)}}"
    properties:
      confluent.metadata.http.auth.credentials.provider: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.id: "{{kafka_connect_replicator_oauth_user}}"
      confluent.metadata.oauthbearer.token.endpoint.url: "{{ oauth_token_uri }}"
  rbac_oauth_secret:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and (not kafka_connect_replicator_mds_cert_auth_only) and (not kafka_connect_replicator_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER
      confluent.metadata.oauthbearer.login.client.secret: "{{kafka_connect_replicator_oauth_password}}"
  rbac_oauth_scope:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and oauth_groups_scope!='none' and (not kafka_connect_replicator_mds_cert_auth_only) and (not kafka_connect_replicator_oauth_client_assertion_enabled) }}"
    properties:
      confluent.metadata.oauthbearer.login.oauth.scope: "{{oauth_groups_scope}}"
  rbac_oauth_third_party_client_assertion:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and kafka_connect_replicator_oauth_client_assertion_enabled and kafka_connect_replicator_oauth_client_assertion_file_base_path !='none' and (not kafka_connect_replicator_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.assertion.file: "{{ kafka_connect_replicator_third_party_oauth_client_assertion_config.mds }}"
  rbac_oauth_local_client_assertion:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and kafka_connect_replicator_oauth_client_assertion_enabled and kafka_connect_replicator_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_replicator_mds_cert_auth_only) }}"
    properties:
      confluent.metadata.oauthbearer.login.credentials.source: OAUTHBEARER_CLIENTASSERTION
      confluent.metadata.oauthbearer.assertion.claim.iss: "{{kafka_connect_replicator_oauth_client_assertion_issuer}}"
      confluent.metadata.oauthbearer.assertion.claim.sub: "{{kafka_connect_replicator_oauth_client_assertion_sub}}"
      confluent.metadata.oauthbearer.assertion.claim.aud: "{{kafka_connect_replicator_oauth_client_assertion_audience}}"
      confluent.metadata.oauthbearer.assertion.private.key.file: "{{ kafka_connect_replicator_oauth_client_assertion_private_key_file_dest_path }}"
  rbac_oauth_local_client_assertion_template_file:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and kafka_connect_replicator_oauth_client_assertion_enabled and kafka_connect_replicator_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_replicator_mds_cert_auth_only) and kafka_connect_replicator_oauth_client_assertion_template_file != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.template.file: "{{ kafka_connect_replicator_oauth_client_assertion_template_file_dest_path }}"
  rbac_oauth_local_client_assertion_passphrase:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and kafka_connect_replicator_oauth_client_assertion_enabled and kafka_connect_replicator_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_replicator_mds_cert_auth_only) and kafka_connect_replicator_oauth_client_assertion_private_key_passphrase != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.private.key.passphrase: "{{kafka_connect_replicator_oauth_client_assertion_private_key_passphrase}}"
  rbac_oauth_local_client_assertion_jti_include:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and kafka_connect_replicator_oauth_client_assertion_enabled and kafka_connect_replicator_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_replicator_mds_cert_auth_only) and kafka_connect_replicator_oauth_client_assertion_jti_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.jti.include: "{{kafka_connect_replicator_oauth_client_assertion_jti_include}}"
  rbac_oauth_local_client_assertion_nbf_include:
    enabled: "{{kafka_connect_replicator_rbac_enabled and oauth_enabled and kafka_connect_replicator_oauth_client_assertion_enabled and kafka_connect_replicator_oauth_client_assertion_file_base_path =='none' and (not kafka_connect_replicator_mds_cert_auth_only) and kafka_connect_replicator_oauth_client_assertion_nbf_include != 'none'}}"
    properties:
      confluent.metadata.oauthbearer.assertion.claim.nbf.include: "{{kafka_connect_replicator_oauth_client_assertion_nbf_include}}"
  kafka_connect_replicator_oauth:
    enabled: "{{kafka_connect_replicator_oauth_enabled}}"
    properties:
      rest.servlet.initializor.classes: io.confluent.common.security.jetty.initializer.AuthenticationHandler
      oauthbearer.jwks.endpoint.url: "{{oauth_jwks_uri}}"
      oauthbearer.sub.claim.name: "{{ oauth_sub_claim }}"
      oauthbearer.expected.issuer: "{{ oauth_issuer_url }}"
  kafka_connect_replicator_oauth_claim:
    enabled: "{{kafka_connect_replicator_oauth_enabled and oauth_groups_claim!='none' }}"
    properties:
      oauthbearer.groups.claim.name: "{{ oauth_groups_claim }}"
  kafka_connect_oauth_audience:
    enabled: "{{kafka_connect_replicator_oauth_enabled and oauth_expected_audience!='none' }}"
    properties:
      oauthbearer.expected.audience: "{{ oauth_expected_audience }}"
  rbac_external_client:
    enabled: "{{kafka_connect_replicator_rbac_enabled and kafka_connect_replicator_erp_tls_enabled }}"
    properties:
      ssl.truststore.location: "{{kafka_connect_replicator_truststore_path}}"
      ssl.truststore.password: "{{kafka_connect_replicator_truststore_storepass}}"
      client.ssl.truststore.location: "{{kafka_connect_replicator_truststore_path}}"
      client.ssl.truststore.password: "{{kafka_connect_replicator_truststore_storepass}}"
  kafka_client:
    enabled: true
    properties: "{{ kafka_connect_replicator_listener | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                '', kafka_connect_replicator_truststore_path, kafka_connect_replicator_truststore_storepass, public_certificates_enabled, kafka_connect_replicator_keystore_path,
                kafka_connect_replicator_keystore_storepass, kafka_connect_replicator_keystore_keypass, false, kafka_connect_replicator_sasl_plain_principal, kafka_connect_replicator_sasl_plain_password,
                kafka_connect_replicator_sasl_scram_principal, kafka_connect_replicator_sasl_scram_password, kafka_connect_replicator_sasl_scram256_principal, kafka_connect_replicator_sasl_scram256_password,
                kerberos_kafka_broker_primary, kafka_connect_replicator_keytab_path,
                kafka_connect_replicator_kerberos_principal|default('kafka'), false, kafka_connect_replicator_ldap_user, kafka_connect_replicator_ldap_password, mds_bootstrap_server_urls,
                oauth_enabled, kafka_connect_replicator_oauth_user, kafka_connect_replicator_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_replicator_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_replicator_third_party_oauth_client_assertion_config.kafka)) }}"
kafka_connect_replicator_combined_properties: "{{kafka_connect_replicator_properties | confluent.platform.combine_properties}}"
kafka_connect_replicator_final_properties: "{{kafka_connect_replicator_combined_properties | combine(kafka_connect_replicator_custom_properties)}}"

kafka_connect_replicator_consumer_properties:
  defaults:
    enabled: true
    properties:
      bootstrap.servers: "{{kafka_connect_replicator_consumer_bootstrap_servers}}"
  rbac_external_client:
    enabled: "{{kafka_connect_replicator_consumer_rbac_enabled and kafka_connect_replicator_consumer_erp_tls_enabled }}"
    properties:
      ssl.truststore.location: "{{kafka_connect_replicator_consumer_truststore_path}}"
      ssl.truststore.password: "{{kafka_connect_replicator_consumer_truststore_storepass}}"
      client.ssl.truststore.location: "{{kafka_connect_replicator_consumer_truststore_path}}"
      client.ssl.truststore.password: "{{kafka_connect_replicator_consumer_truststore_storepass}}"
  kafka_client:
    enabled: true
    properties: "{{ kafka_connect_replicator_consumer_listener | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                '', kafka_connect_replicator_truststore_path, kafka_connect_replicator_truststore_storepass, public_certificates_enabled, kafka_connect_replicator_keystore_path,
                kafka_connect_replicator_keystore_storepass, kafka_connect_replicator_keystore_keypass, false, kafka_connect_replicator_consumer_sasl_plain_principal, kafka_connect_replicator_consumer_sasl_plain_password,
                kafka_connect_replicator_consumer_sasl_scram_principal, kafka_connect_replicator_consumer_sasl_scram_password, kafka_connect_replicator_consumer_sasl_scram256_principal, kafka_connect_replicator_consumer_sasl_scram256_password,
                kerberos_kafka_broker_primary, kafka_connect_replicator_keytab_path,
                kafka_connect_replicator_kerberos_principal|default('kafka'), false, kafka_connect_replicator_ldap_user, kafka_connect_replicator_ldap_password, mds_bootstrap_server_urls,
                oauth_enabled, kafka_connect_replicator_consumer_oauth_user, kafka_connect_replicator_consumer_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_replicator_consumer_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_replicator_consumer_third_party_oauth_client_assertion_config.kafka)) }}"
kafka_connect_replicator_consumer_combined_properties: "{{kafka_connect_replicator_consumer_properties | confluent.platform.combine_properties}}"
kafka_connect_replicator_consumer_final_properties: "{{kafka_connect_replicator_consumer_combined_properties | combine(kafka_connect_replicator_consumer_custom_properties)}}"

kafka_connect_replicator_producer_properties:
  defaults:
    enabled: true
    properties:
      bootstrap.servers: "{{kafka_connect_replicator_producer_bootstrap_servers}}"
  rbac_external_client:
    enabled: "{{kafka_connect_replicator_producer_rbac_enabled and kafka_connect_replicator_producer_erp_tls_enabled }}"
    properties:
      ssl.truststore.location: "{{kafka_connect_replicator_producer_truststore_path}}"
      ssl.truststore.password: "{{kafka_connect_replicator_producer_truststore_storepass}}"
      client.ssl.truststore.location: "{{kafka_connect_replicator_producer_truststore_path}}"
      client.ssl.truststore.password: "{{kafka_connect_replicator_producer_truststore_storepass}}"
  kafka_client:
    enabled: true
    properties: "{{ kafka_connect_replicator_producer_listener | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                '', kafka_connect_replicator_truststore_path, kafka_connect_replicator_truststore_storepass, public_certificates_enabled, kafka_connect_replicator_keystore_path,
                kafka_connect_replicator_keystore_storepass, kafka_connect_replicator_keystore_keypass, false, kafka_connect_replicator_producer_sasl_plain_principal, kafka_connect_replicator_producer_sasl_plain_password,
                kafka_connect_replicator_producer_sasl_scram_principal, kafka_connect_replicator_producer_sasl_scram_password, kafka_connect_replicator_producer_sasl_scram256_principal, kafka_connect_replicator_producer_sasl_scram256_password,
                kerberos_kafka_broker_primary, kafka_connect_replicator_keytab_path,
                kafka_connect_replicator_kerberos_principal|default('kafka'), false, kafka_connect_replicator_ldap_user, kafka_connect_replicator_ldap_password, mds_bootstrap_server_urls,
                oauth_enabled, kafka_connect_replicator_producer_oauth_user, kafka_connect_replicator_producer_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_replicator_producer_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_replicator_producer_third_party_oauth_client_assertion_config.kafka)) }}"
kafka_connect_replicator_producer_combined_properties: "{{kafka_connect_replicator_producer_properties | confluent.platform.combine_properties}}"
kafka_connect_replicator_producer_final_properties: "{{kafka_connect_replicator_producer_combined_properties | combine(kafka_connect_replicator_producer_custom_properties)}}"

kafka_connect_replicator_monitoring_interceptor_properties:
  defaults:
    enabled: "{{kafka_connect_replicator_monitoring_interceptors_enabled|bool}}"
    properties:
      confluent.monitoring.interceptor.topic: _confluent-monitoring
      producer.confluent.monitoring.interceptor.bootstrap.servers: "{{kafka_connect_replicator_monitoring_interceptor_bootstrap_servers}}"
      consumer.confluent.monitoring.interceptor.bootstrap.servers: "{{kafka_connect_replicator_monitoring_interceptor_bootstrap_servers}}"
  rbac_external_client:
    enabled: "{{kafka_connect_replicator_monitoring_interceptor_rbac_enabled and kafka_connect_replicator_monitoring_interceptor_erp_tls_enabled }}"
    properties:
      ssl.truststore.location: "{{kafka_connect_replicator_monitoring_interceptor_truststore_path}}"
      ssl.truststore.password: "{{kafka_connect_replicator_monitoring_interceptor_truststore_storepass}}"
      client.ssl.truststore.location: "{{kafka_connect_replicator_monitoring_interceptor_truststore_path}}"
      client.ssl.truststore.password: "{{kafka_connect_replicator_monitoring_interceptor_truststore_storepass}}"
  kafka_client:
    enabled: true
    properties: "{{ kafka_connect_replicator_monitoring_interceptor_listener | confluent.platform.client_properties(ssl_enabled, False, ssl_mutual_auth_enabled, sasl_protocol,
                '', kafka_connect_replicator_truststore_path, kafka_connect_replicator_truststore_storepass, public_certificates_enabled, kafka_connect_replicator_keystore_path,
                kafka_connect_replicator_keystore_storepass, kafka_connect_replicator_keystore_keypass, false, kafka_connect_replicator_monitoring_interceptor_sasl_plain_principal, kafka_connect_replicator_monitoring_interceptor_sasl_plain_password,
                kafka_connect_replicator_monitoring_interceptor_sasl_scram_principal, kafka_connect_replicator_monitoring_interceptor_sasl_scram_password, kafka_connect_replicator_monitoring_interceptor_sasl_scram256_principal, kafka_connect_replicator_monitoring_interceptor_sasl_scram256_password,
                kerberos_kafka_broker_primary, kafka_connect_replicator_keytab_path,
                kafka_connect_replicator_kerberos_principal|default('kafka'), false, kafka_connect_replicator_ldap_user, kafka_connect_replicator_ldap_password, mds_bootstrap_server_urls,
                oauth_enabled, kafka_connect_replicator_monitoring_interceptor_oauth_user, kafka_connect_replicator_monitoring_interceptor_oauth_password, oauth_groups_scope, oauth_token_uri, idp_self_signed, false, kafka_connect_replicator_monitoring_interceptor_oauth_client_assertion_config | confluent.platform.replace_client_assertion_file(kafka_connect_replicator_third_party_oauth_client_assertion_config.kafka)) }}"
kafka_connect_replicator_monitoring_interceptor_combined_properties: "{{kafka_connect_replicator_monitoring_interceptor_properties | confluent.platform.combine_properties}}"
kafka_connect_replicator_monitoring_interceptor_final_properties: "{{kafka_connect_replicator_monitoring_interceptor_combined_properties | combine(kafka_connect_replicator_monitoring_interceptor_custom_properties)}}"

#### USM Agent Variables ####
usm_agent_service_name: usm-agent
usm_agent_default_user: cp-usm-agent
usm_agent_default_group: confluent
confluent_usm_agent_archive_version: "{{confluent_usm_agent_full_package_version}}"
usm_agent_binary_base_path: "{{ ((config_base_path,('confluent-usm-agent-',confluent_usm_agent_archive_version) | join) | path_join) if installation_method == 'archive' else '/usr' }}"
usm_agent_config_prefix_path: "{{ usm_agent_config_prefix.strip('/') }}"

usm_agent:
  config_file: "{{ ((config_base_path,('confluent-usm-agent-',confluent_usm_agent_archive_version) | join) | path_join) if installation_method == 'archive' else '' }}/etc/confluent/usm-agent/usm-agent.properties"
  systemd_file: "{{ (systemd_base_dir, (usm_agent_service_name, '.service') | join) | path_join }}"
  systemd_override: /etc/systemd/system/{{usm_agent_service_name}}.service.d/override.conf
  ccloud_credential_file: "{{ ((config_base_path,('confluent-usm-agent-',confluent_usm_agent_archive_version) | join) | path_join) if installation_method == 'archive' else '' }}/etc/confluent/usm-agent/secrets/ccloud_credential.yaml"
  cp_credential_file: "{{ ((config_base_path,('confluent-usm-agent-',confluent_usm_agent_archive_version) | join) | path_join) if installation_method == 'archive' else '' }}/etc/confluent/usm-agent/secrets/cp_credential.yaml"
  log_dir: "/var/log/confluent/usm-agent"


usm_agent_ccloud_environment_id: "{{ ccloud_environment_id }}"
usm_agent_ccloud_host: "{{ ccloud_endpoint | urlsplit('hostname') }}"
usm_agent_ccloud_port: "{{ ccloud_endpoint | urlsplit('port') | default(443, true) }}"
usm_agent_ccloud_ssl_enabled: true

usm_agent_ccloud_credential:
  username: "{{ ccloud_credential.username }}"
  password: "{{ ccloud_credential.password }}"

usm_agent_secrets_dir: "{{ (config_base_path, usm_agent_config_prefix_path, 'secrets') | path_join }}"

usm_agent_ccloud_basic_auth_encoded: "{{ (usm_agent_ccloud_credential.username + ':' + usm_agent_ccloud_credential.password) | b64encode }}"

usm_agent_properties:
  defaults:
    enabled: true
    properties:
      confluent.usm-agent.listener.admin.port: "{{ usm_agent_admin_port }}"
      confluent.usm-agent.listener.controlplane.port: "{{ usm_agent_controlplane_port }}"
      confluent.usm-agent.listener.dataplane.port: "{{ usm_agent_dataplane_port }}"
      confluent.usm-agent.listener.dataplane.basic.auth.enabled: "{{ usm_agent_basic_auth_enabled|string|lower }}"
      confluent.usm-agent.listener.dataplane.ssl.enabled: "{{ usm_agent_ssl_enabled|string|lower }}"
      confluent.usm-agent.listener.dataplane.ssl.client.auth.enabled: "{{ usm_agent_ssl_mutual_auth_enabled|string|lower }}"
      confluent.usm-agent.ccloud.environment-id: "{{ usm_agent_ccloud_environment_id }}"
      confluent.usm-agent.ccloud.host: "{{ usm_agent_ccloud_host }}"
      confluent.usm-agent.ccloud.port: "{{ usm_agent_ccloud_port }}"
      confluent.usm-agent.ccloud.basic.auth.credential.location: "{{ usm_agent.ccloud_credential_file }}"
      confluent.usm-agent.ccloud.ssl.enabled: "{{ usm_agent_ccloud_ssl_enabled|string|lower }}"
      confluent.usm-agent.ccloud.ssl.trusted.ca.location: "{{ usm_agent_ccloud_ssl_trusted_ca_location }}"
      confluent.usm-agent.listener.monitoring.port: "{{ usm_agent_listener_monitoring_port }}"
  basic:
    enabled: "{{ usm_agent_basic_auth_enabled }}"
    properties:
      confluent.usm-agent.listener.dataplane.basic.auth.credential.location: "{{ usm_agent.cp_credential_file }}"
  ssl:
    enabled: "{{ usm_agent_ssl_enabled }}"
    properties:
      confluent.usm-agent.listener.dataplane.ssl.server.certificate.location: "{{ usm_agent_cert_path }}"
      confluent.usm-agent.listener.dataplane.ssl.server.key.location: "{{ usm_agent_key_path }}"
      confluent.usm-agent.listener.dataplane.ssl.server.key.password.location: "{{ usm_agent_key_password_file }}"
  mtls:
    enabled: "{{ usm_agent_ssl_mutual_auth_enabled }}"
    properties:
      confluent.usm-agent.listener.dataplane.ssl.client.auth.trusted.ca.location: "{{ usm_agent_ca_cert_path }}"


usm_agent_combined_properties: "{{usm_agent_properties | confluent.platform.combine_properties}}"

usm_agent_final_properties: "{{ usm_agent_combined_properties | combine(usm_agent_custom_properties) }}"