upgrade_kafka_broker_log_format.yml

---
- name: Kafka Broker Upgrade - Set Log Message Format Version to Current Version
  hosts: kafka_broker
  serial: 1
  vars:
    kafka_broker_upgrade_end_version: "6.1"
  vars_files:
    - roles/confluent.kafka_broker/defaults/main.yml
  handlers:
    - name: restart kafka
      include_tasks: roles/confluent.kafka_broker/tasks/restart_and_wait.yml
  tasks:
    - import_role:
        name: confluent.variables

    - name: Set confluent_server_enabled Variable
      include_tasks: tasks/set_confluent_server_enabled.yml

    - name: Load server.properties
      slurp:
        src: "{{ kafka_broker.config_file }}"
      register: slurped_properties

    - set_fact:
        kafka_broker_rest_proxy_enabled: "{{ True if (slurped_properties.content|b64decode).split('\n') | select('match', '^kafka.rest.enable=true') | list | length == 1 else False }}"

    - set_fact:
        log_message_format_version: "{{'2.3' if kafka_broker_upgrade_end_version is match('5.3.*') else
          '2.4' if kafka_broker_upgrade_end_version is match('5.4.*') else
          '2.5' if kafka_broker_upgrade_end_version is match('5.5.*') else
          '2.6' if kafka_broker_upgrade_end_version is match('6.0.*') else
          '2.7' }}"

    - name: Set Log Message Format Version to Current Version
      lineinfile:
        name: "{{kafka_broker.config_file}}"
        line: "log.message.format.version={{log_message_format_version}}"
        regexp: log.message.format.version.*
        state: present
      notify: restart kafka

    - meta: flush_handlers

    - name: Kafka Broker Health Check
      include_tasks: roles/confluent.kafka_broker/tasks/health_check.yml
      when: not ansible_check_mode